Connect Kube gateway part 1: lib/teleterm/gateway

integration/proxy/proxy_helpers.go

@@ -563,7 +563,9 @@ 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.NewKubeForwardProxy(alpnproxy.KubeForwardProxyConfig{
+		ForwardAddr: lpAddr,
+	})
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		fp.Close()

lib/srv/alpnproxy/forward_proxy.go

@@ -100,7 +100,7 @@ func (p *ForwardProxy) Start() error {
 
 // Close closes the forward proxy.
 func (p *ForwardProxy) Close() error {
-	if err := p.cfg.Listener.Close(); err != nil {
+	if err := p.cfg.Listener.Close(); err != nil && !utils.IsUseOfClosedNetworkError(err) {
 		return trace.Wrap(err)
 	}
 	return nil

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"
@@ -270,30 +273,93 @@ func NewKubeListener(casByTeleportCluster map[string]tls.Certificate) (net.Liste
 	return listener, trace.Wrap(err)
 }
 
-// NewKubeForwardProxy creates a forward proxy for kube access.
-func NewKubeForwardProxy(ctx context.Context, listenPort, forwardAddr string) (*ForwardProxy, error) {
-	listenAddr := "localhost:0"
-	if listenPort != "" {
-		listenAddr = "localhost:" + listenPort
+// KubeForwardProxyConfig is the config for making kube forward proxy.
+type KubeForwardProxyConfig struct {
+	// CloseContext is the close context.
+	CloseContext context.Context
+	// ListenPort is the localhost port to listen.
+	ListenPort string
+	// Listener is the listener for the forward proxy. A listener is created
+	// from ListenPort if Listener is not provided.
+	Listener net.Listener
+	// ForwardAddr is the target address the requests get forwarded to.
+	ForwardAddr string
+}
+
+// CheckAndSetDefaults checks and sets default config values.
+func (c *KubeForwardProxyConfig) CheckAndSetDefaults() error {
+	if c.ForwardAddr == "" {
+		return trace.BadParameter("missing forward address")
+	}
+	if c.CloseContext == nil {
+		c.CloseContext = context.Background()
 	}
+	if c.Listener == nil {
+		if c.ListenPort == "" {
+			c.ListenPort = "0"
+		}
 
-	listener, err := net.Listen("tcp", listenAddr)
-	if err != nil {
-		return nil, trace.Wrap(err)
+		listener, err := net.Listen("tcp", "localhost:"+c.ListenPort)
+		if err != nil {
+			return trace.Wrap(err)
+		}
+		c.Listener = listener
 	}
+	return nil
+}
 
+// NewKubeForwardProxy creates a forward proxy for kube access.
+func NewKubeForwardProxy(config KubeForwardProxyConfig) (*ForwardProxy, error) {
+	if err := config.CheckAndSetDefaults(); err != nil {
+		return nil, trace.Wrap(err)
+	}
 	fp, err := NewForwardProxy(ForwardProxyConfig{
-		Listener:     listener,
-		CloseContext: ctx,
+		Listener:     config.Listener,
+		CloseContext: config.CloseContext,
 		Handlers: []ConnectRequestHandler{
 			NewForwardToHostHandler(ForwardToHostHandlerConfig{
 				MatchFunc: MatchAllRequests,
-				Host:      forwardAddr,
+				Host:      config.ForwardAddr,
 			}),
 		},
 	})
 	if err != nil {
-		return nil, trace.NewAggregate(listener.Close(), err)
+		return nil, trace.NewAggregate(config.Listener.Close(), err)
 	}
 	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/srv/alpnproxy/local_proxy.go

@@ -161,6 +161,14 @@ func NewLocalProxy(cfg LocalProxyConfig, opts ...LocalProxyConfigOpt) (*LocalPro
 
 // Start starts the LocalProxy.
 func (l *LocalProxy) Start(ctx context.Context) error {
+	if l.cfg.HTTPMiddleware != nil {
+		return trace.Wrap(l.StartHTTPAccessProxy(ctx))
+	}
+	return trace.Wrap(l.start(ctx))
+}
+
+// start starts the LocalProxy for raw TCP or raw TLS (non-HTTP) connections.
+func (l *LocalProxy) start(ctx context.Context) error {
 	if l.cfg.Middleware != nil {
 		err := l.cfg.Middleware.OnStart(ctx, l)
 		if err != nil {
@@ -238,7 +246,7 @@ func (l *LocalProxy) handleDownstreamConnection(ctx context.Context, downstreamC
 func (l *LocalProxy) Close() error {
 	l.cancel()
 	if l.cfg.Listener != nil {
-		if err := l.cfg.Listener.Close(); err != nil {
+		if err := l.cfg.Listener.Close(); err != nil && !utils.IsUseOfClosedNetworkError(err) {
 			return trace.Wrap(err)
 		}
 	}
@@ -265,6 +273,9 @@ func (l *LocalProxy) makeHTTPReverseProxy(certs []tls.Certificate) *httputil.Rev
 			outReq.URL.Host = l.cfg.RemoteProxyAddr
 		},
 		ModifyResponse: func(response *http.Response) error {
+			// Ask the client to close the connection to avoid re-use.
+			response.Header.Add("Connection", "close")
+
 			errHeader := response.Header.Get(commonApp.TeleportAPIErrorHeader)
 			if errHeader != "" {
 				// TODO: find a cleaner way of formatting the error.

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
 }
+
+// IsDB returns true if URI is a database resource.
+func (r ResourceURI) IsDB() bool {
+	return r.GetDbName() != ""
+}
+
+// IsKube returns true if URI is a kube resource.
+func (r ResourceURI) IsKube() bool {
+	return r.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
@@ -216,3 +262,67 @@ func TestGetRootClusterURI(t *testing.T) {
 		})
 	}
 }
+
+func TestIsDB(t *testing.T) {
+	tests := []struct {
+		in    uri.ResourceURI
+		check require.BoolAssertionFunc
+	}{
+		{
+			in:    uri.NewClusterURI("foo").AppendDB("db"),
+			check: require.True,
+		},
+		{
+			in:    uri.NewClusterURI("foo").AppendLeafCluster("bar").AppendDB("db"),
+			check: require.True,
+		},
+		{
+			in:    uri.NewClusterURI("foo"),
+			check: require.False,
+		},
+		{
+			in:    uri.NewClusterURI("foo").AppendLeafCluster("bar"),
+			check: require.False,
+		},
+		{
+			in:    uri.NewClusterURI("foo").AppendKube("kube"),
+			check: require.False,
+		},
+	}
+
+	for _, tt := range tests {
+		tt.check(t, tt.in.IsDB())
+	}
+}
+
+func TestIsKube(t *testing.T) {
+	tests := []struct {
+		in    uri.ResourceURI
+		check require.BoolAssertionFunc
+	}{
+		{
+			in:    uri.NewClusterURI("foo").AppendKube("kube"),
+			check: require.True,
+		},
+		{
+			in:    uri.NewClusterURI("foo").AppendLeafCluster("bar").AppendKube("kube"),
+			check: require.True,
+		},
+		{
+			in:    uri.NewClusterURI("foo"),
+			check: require.False,
+		},
+		{
+			in:    uri.NewClusterURI("foo").AppendLeafCluster("bar"),
+			check: require.False,
+		},
+		{
+			in:    uri.NewClusterURI("foo").AppendDB("db"),
+			check: require.False,
+		},
+	}
+
+	for _, tt := range tests {
+		tt.check(t, tt.in.IsKube())
+	}
+}