Kubernetes forwarding in the Relay Service (#60974)

* Add signaling for supported tunnel types to relay tunnel

* Enable relay tunnel client for kubernetes_service

* Add kube_server to the relay cache

* Add passive forwarder for kubernetes access through the relay

* Add wildcard SNI suffix to kube service SANs if a relay is configured

* Add SNI-dispatching grpc transport for the relay transport server

* Enable kubernetes forwarding in the relay service

* Advertise support for kube tunnels in the relay

* Add tests

gen/proto/go/teleport/relaytunnel/v1alpha/discovery_service.pb.go

@@ -79,8 +79,11 @@ type DiscoverResponse struct {
 	state                 protoimpl.MessageState `protogen:"open.v1"`
 	RelayGroup            string                 `protobuf:"bytes,1,opt,name=relay_group,json=relayGroup,proto3" json:"relay_group,omitempty"`
 	TargetConnectionCount int32                  `protobuf:"varint,2,opt,name=target_connection_count,json=targetConnectionCount,proto3" json:"target_connection_count,omitempty"`
-	unknownFields         protoimpl.UnknownFields
-	sizeCache             protoimpl.SizeCache
+	// the list of api/types.TunnelTypes supported by this Relay group; if empty,
+	// it should be treated as containing only types.NodeTunnel ("node")
+	SupportedTunnelTypes []string `protobuf:"bytes,3,rep,name=supported_tunnel_types,json=supportedTunnelTypes,proto3" json:"supported_tunnel_types,omitempty"`
+	unknownFields        protoimpl.UnknownFields
+	sizeCache            protoimpl.SizeCache
 }
 
 func (x *DiscoverResponse) Reset() {
@@ -127,16 +130,24 @@ func (x *DiscoverResponse) GetTargetConnectionCount() int32 {
 	return 0
 }
 
+func (x *DiscoverResponse) GetSupportedTunnelTypes() []string {
+	if x != nil {
+		return x.SupportedTunnelTypes
+	}
+	return nil
+}
+
 var File_teleport_relaytunnel_v1alpha_discovery_service_proto protoreflect.FileDescriptor
 
 const file_teleport_relaytunnel_v1alpha_discovery_service_proto_rawDesc = "" +
 	"\n" +
 	"4teleport/relaytunnel/v1alpha/discovery_service.proto\x12\x1cteleport.relaytunnel.v1alpha\"\x11\n" +
-	"\x0fDiscoverRequest\"k\n" +
+	"\x0fDiscoverRequest\"\xa1\x01\n" +
 	"\x10DiscoverResponse\x12\x1f\n" +
 	"\vrelay_group\x18\x01 \x01(\tR\n" +
 	"relayGroup\x126\n" +
-	"\x17target_connection_count\x18\x02 \x01(\x05R\x15targetConnectionCount2}\n" +
+	"\x17target_connection_count\x18\x02 \x01(\x05R\x15targetConnectionCount\x124\n" +
+	"\x16supported_tunnel_types\x18\x03 \x03(\tR\x14supportedTunnelTypes2}\n" +
 	"\x10DiscoveryService\x12i\n" +
 	"\bDiscover\x12-.teleport.relaytunnel.v1alpha.DiscoverRequest\x1a..teleport.relaytunnel.v1alpha.DiscoverResponseB`Z^github.com/gravitational/teleport/gen/proto/go/teleport/relaytunnel/v1alpha;relaytunnelv1alphab\x06proto3"
 

lib/auth/authclient/api.go

@@ -390,6 +390,8 @@ type ReadRelayAccessPoint interface {
 	GetSessionRecordingConfig(ctx context.Context) (types.SessionRecordingConfig, error)
 
 	GetUser(ctx context.Context, name string, withSecrets bool) (types.User, error)
+
+	GetKubernetesServers(ctx context.Context) ([]types.KubeServer, error)
 }
 
 // ReadRemoteProxyAccessPoint is a read only API interface implemented by a certificate authority (CA) to be

lib/cache/cache.go

@@ -285,6 +285,7 @@ func ForRelay(cfg Config) Config {
 		{Kind: types.KindCertAuthority, Filter: makeAllKnownCAsFilter().IntoMap()},
 		{Kind: types.KindClusterAuthPreference},
 		{Kind: types.KindClusterNetworkingConfig},
+		{Kind: types.KindKubeServer},
 		{Kind: types.KindNode},
 		{Kind: types.KindRelayServer},
 		{Kind: types.KindRole},

lib/kube/proxy/server.go

@@ -47,6 +47,7 @@ import (
 	"github.com/gravitational/teleport/lib/labels"
 	"github.com/gravitational/teleport/lib/limiter"
 	"github.com/gravitational/teleport/lib/multiplexer"
+	"github.com/gravitational/teleport/lib/relaytunnel"
 	"github.com/gravitational/teleport/lib/reversetunnelclient"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/services/readonly"
@@ -72,6 +73,9 @@ type TLSServerConfig struct {
 	GetRotation services.RotationGetter
 	// ConnectedProxyGetter gets the proxies teleport is connected to.
 	ConnectedProxyGetter reversetunnelclient.ConnectedProxyGetter
+	// RelayInfoGetter is the function used to get the relay tunnel client info
+	// to fill in the server heartbeats.
+	RelayInfoGetter relaytunnel.GetRelayInfoFunc
 	// Log is the logger.
 	Log *slog.Logger
 	// Selectors is a list of resource monitor selectors.
@@ -504,18 +508,27 @@ func (t *TLSServer) GetServerInfo(name string) (*types.KubernetesServerV3, error
 		name += teleport.KubeLegacyProxySuffix
 	}
 
+	var relayGroup string
+	var relayIDs []string
+	if t.RelayInfoGetter != nil {
+		// relayInfoGetter returns a copy of the slice, so we can move it in the
+		// protobuf message
+		relayGroup, relayIDs = t.RelayInfoGetter()
+	}
 	srv, err := types.NewKubernetesServerV3(
 		types.Metadata{
 			Name:      name,
 			Namespace: t.Namespace,
 		},
 		types.KubernetesServerSpecV3{
-			Version:  teleport.Version,
-			Hostname: addr,
-			HostID:   t.TLSServerConfig.HostID,
-			Rotation: t.getRotationState(),
-			Cluster:  cluster,
-			ProxyIDs: t.ConnectedProxyGetter.GetProxyIDs(),
+			Version:    teleport.Version,
+			Hostname:   addr,
+			HostID:     t.TLSServerConfig.HostID,
+			Rotation:   t.getRotationState(),
+			Cluster:    cluster,
+			ProxyIDs:   t.ConnectedProxyGetter.GetProxyIDs(),
+			RelayGroup: relayGroup,
+			RelayIds:   relayIDs,
 		},
 	)
 	if err != nil {

lib/kube/relay/hash.go

@@ -0,0 +1,54 @@
+// Teleport
+// Copyright (C) 2025 Gravitational, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package relay
+
+import (
+	"crypto/sha256"
+	"encoding/base32"
+)
+
+// base32hex is the "Extended Hex Alphabet" defined in RFC 4648 but with
+// lowercase letters and no padding.
+var base32hex = base32.NewEncoding("0123456789abcdefghijklmnopqrstuv").WithPadding(base32.NoPadding)
+
+const (
+	hashLen = sha256.Size
+	// encodedHashLen is the length of the base32 encoding of [hashLen] bytes,
+	// without padding.
+	encodedHashLen = hashLen/5*8 + (hashLen%5*8+4)/5
+)
+
+func hashForTarget(teleportClusterName, kubeClusterName string) [hashLen]byte {
+	h := sha256.New()
+	var buf [hashLen]byte
+	buf = sha256.Sum256([]byte(teleportClusterName))
+	h.Write(buf[:])
+	buf = sha256.Sum256([]byte(kubeClusterName))
+	h.Write(buf[:])
+	h.Sum(buf[:0])
+	return buf
+}
+
+// SNILabelForKubeCluster returns the domain label used in front of [SNISuffix]
+// to identify a Kubernetes cluster as the target for a passively routed Relay
+// connection. It consists of [SNIPrefixForKubeCluster] ("cluster-") followed by
+// the base32hex encoding of the SHA256 hash of Teleport cluster name and
+// Kubernetes cluster name, for a total of 60 ASCII lowercase characters.
+func SNILabelForKubeCluster(teleportClusterName, kubeClusterName string) string {
+	h := hashForTarget(teleportClusterName, kubeClusterName)
+	return SNIPrefixForKubeCluster + base32hex.EncodeToString(h[:])
+}

lib/kube/relay/hash_test.go

@@ -0,0 +1,69 @@
+// Teleport
+// Copyright (C) 2025 Gravitational, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package relay
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestEncodedHashLen(t *testing.T) {
+	//nolint:staticcheck // the explicit type in the declaration is the entire
+	// point of this statement
+	var _ [hashLen]byte = hashForTarget("", "")
+
+	//nolint:testifylint // the parameter order is deliberate,
+	// b32hex.EncodedLen(hashLen) is the source of truth and we are testing the
+	// value of the const
+	require.Equal(t, base32hex.EncodedLen(hashLen), encodedHashLen)
+}
+
+func TestHashForTarget(t *testing.T) {
+	testCases := []struct {
+		teleportClusterName    string
+		kubeClusterName        string
+		hash                   string
+		sniLabelForKubeCluster string
+	}{
+		{
+			teleportClusterName:    "",
+			kubeClusterName:        "",
+			hash:                   "-\xba]\xbc3\x9es\x16\xae\xa2h?\xaf\x83\x9c\x1b{\x1e\xe21=\xb7\x92\x11%\x88\x11\x8d\xf0f\xaa5",
+			sniLabelForKubeCluster: "cluster-5mt5rf1jjpphdbl2d0vqv0ss3dthtohh7mrp4495h08ors36l8qg",
+		},
+		{
+			teleportClusterName:    "teleport.example.com",
+			kubeClusterName:        "prod-1",
+			hash:                   "%0\x86\x19\x99\x17\x97\xf9uX~\x00\xe4?\xf4aPO\v\xde%gq\xc7\xdf\xc6\x05|\xd1y\xd7r",
+			sniLabelForKubeCluster: "cluster-4ko8c6cp2ubvitaofo0e8fvkc584u2uu4ljn3huvoo2npkbpqtp0",
+		},
+
+		{
+			teleportClusterName:    "8701ae5eecf94a25adc7a90e3f373a25",
+			kubeClusterName:        "3929312bc0c74cd491a882f7abd7edbd",
+			hash:                   "\x0f\x12\xae\xf8\x83\xc2\xd3 j-\xe9\x87\xea\xdf\x18s\xd7\f\xb6\xcf\xcd\xce\x17\x89\x19c\r\xa2\x99\xd8A\xdd",
+			sniLabelForKubeCluster: "cluster-1s9atu43ob9i0qhdt63ulnooefbgpdmfpn71f28pcc6q56eo87eg",
+		},
+	}
+
+	for _, tc := range testCases {
+		hft := hashForTarget(tc.teleportClusterName, tc.kubeClusterName)
+		require.Equal(t, tc.hash, string(hft[:]))
+		require.Equal(t, tc.sniLabelForKubeCluster, SNILabelForKubeCluster(tc.teleportClusterName, tc.kubeClusterName))
+	}
+}

lib/kube/relay/passive_forwarder.go

@@ -0,0 +1,407 @@
+// Teleport
+// Copyright (C) 2025 Gravitational, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package relay
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"log/slog"
+	"net"
+	"slices"
+	"strings"
+	"sync"
+
+	"github.com/gravitational/trace"
+	lru "github.com/hashicorp/golang-lru/v2"
+
+	apiconstants "github.com/gravitational/teleport/api/constants"
+	apitypes "github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/healthcheck"
+	"github.com/gravitational/teleport/lib/services/readonly"
+	"github.com/gravitational/teleport/lib/utils"
+	logutils "github.com/gravitational/teleport/lib/utils/log"
+)
+
+// SNISuffix is the suffix of the server name that signals that a connection is
+// intended for Kubernetes Access through a relay. The format of the domain
+// label in front of SNISuffix depends on the desired target for the connection.
+// For now, the only supported class of targets is identified by a prefix of
+// [SNIPrefixForKubeCluster] ("cluster-") to signify a connection for a given
+// Kubernetes cluster by name.
+const SNISuffix = ".kuberelay." + apiconstants.APIDomain
+
+// SNIPrefixForKubeCluster is the prefix of the domain label used to identify a
+// Kubernetes cluster as the target for a passively routed Relay connection.
+const SNIPrefixForKubeCluster = "cluster-"
+
+// When adding more target types (for example "session-") keep in mind that the
+// entire reason for the hashing scheme in [SNILabelForKubeCluster] is that
+// individual labels are limited to 63 characters (and the total FQDN is limited
+// to 253 characters, but that's not relevant for us since Kube agents can only
+// have one wildcard label in front of [SNISuffix]), so mind the total length of
+// the data you need to encode.
+
+type RelayTunnelDialFunc = func(ctx context.Context, hostID string, tunnelType apitypes.TunnelType, src, dst net.Addr) (net.Conn, error)
+type RelayPeerDialFunc = func(ctx context.Context, hostID string, tunnelType apitypes.TunnelType, relayIDs []string, src, dst net.Addr) (net.Conn, error)
+
+type GetKubeServersWithFilterFunc = func(ctx context.Context, filter func(readonly.KubeServer) bool) ([]apitypes.KubeServer, error)
+
+// PassiveForwarderConfig contains the parameters for [NewPassiveForwarder].
+type PassiveForwarderConfig struct {
+	Log *slog.Logger
+
+	ClusterName string
+	GroupName   string
+
+	// GetKubeServersWithFilter should return the Kube server resources known to
+	// the cluster that pass the given filter function.
+	GetKubeServersWithFilter GetKubeServersWithFilterFunc
+
+	// LocalDial should dial the given target if it's available as a reverse
+	// tunnel attached to the local instance.
+	LocalDial RelayTunnelDialFunc
+	// PeerDial should dial the given target through a known list of relay IDs.
+	// It will only be called for targets that are advertising the same relay
+	// group, but the function should check that any given relay in the list
+	// belongs to the correct group before attempting to use it.
+	PeerDial RelayPeerDialFunc
+}
+
+// NewPassiveForwarder returns a [PassiveForwarder] with the given config.
+func NewPassiveForwarder(cfg PassiveForwarderConfig) (*PassiveForwarder, error) {
+	if cfg.Log == nil {
+		return nil, trace.BadParameter("missing Log")
+	}
+	if cfg.ClusterName == "" {
+		return nil, trace.BadParameter("missing ClusterName")
+	}
+	if cfg.GroupName == "" {
+		return nil, trace.BadParameter("missing GroupName")
+	}
+	if cfg.GetKubeServersWithFilter == nil {
+		return nil, trace.BadParameter("missing GetKubeServersWithFilter")
+	}
+	if cfg.LocalDial == nil {
+		return nil, trace.BadParameter("missing LocalDial")
+	}
+	if cfg.PeerDial == nil {
+		return nil, trace.BadParameter("missing PeerDial")
+	}
+
+	resolvedNames, err := lru.New[[hashLen]byte, string](64)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	// This context is owned by PassiveForwarder and it's used for operations
+	// that get synchronously canceled by Close.
+	ctx, cancel := context.WithCancel(context.Background())
+
+	return &PassiveForwarder{
+		log: cfg.Log,
+
+		clusterName: cfg.ClusterName,
+		groupName:   cfg.GroupName,
+
+		getKubeServersWithFilter: cfg.GetKubeServersWithFilter,
+
+		localDial: cfg.LocalDial,
+		peerDial:  cfg.PeerDial,
+
+		resolvedNames: resolvedNames,
+
+		ctx:    ctx,
+		cancel: cancel,
+	}, nil
+}
+
+// PassiveForwarder implements the logic to route and forward connections to
+// Kubernetes Service agents connected to a Relay.
+type PassiveForwarder struct {
+	log *slog.Logger
+
+	clusterName string
+	groupName   string
+
+	getKubeServersWithFilter GetKubeServersWithFilterFunc
+
+	localDial RelayTunnelDialFunc
+	peerDial  RelayPeerDialFunc
+
+	// resolvedNames contains (hash, name) pairs for which
+	// hash == hashForTarget(clusterName, name); pairs are only _added_ to the
+	// LRU if v is a kube cluster served by an agent connected to this relay
+	// group, but that doesn't guarantee that the cluster is still being served
+	// if the pair is looked up at a later time.
+	resolvedNames *lru.Cache[[hashLen]byte, string]
+
+	// ctx is used to signal that long running operations should unblock and
+	// return, especially operations that have added to wg because Close should
+	// wait until their end. It's owned by PassiveForwarder and should be
+	// canceled by calling Close; it does not belong to a context tree and does
+	// not outlive any call with a parent context. The intended operation of
+	// this object is almost entirely based on I/O and explicit closing, so
+	// there's no natural way to receive contexts associated with individual
+	// operations triggered by callers of its methods.
+	ctx    context.Context
+	cancel context.CancelFunc
+
+	mu sync.Mutex
+	// wg synchronizes with ctx and mu: wg.Add must only be called while holding
+	// mu if ctx.Err == nil, wg.Wait must only be called after cancel is called
+	// while holding mu.
+	wg sync.WaitGroup
+}
+
+// Close closes all ongoing connections and blocks until the associated
+// resources are gone.
+func (p *PassiveForwarder) Close() error {
+	p.mu.Lock()
+	// we cancel the context while holding the lock to signal that nothing
+	// long-running should begin in the background, so nothing will add to wg
+	// and we can wait on it
+	p.cancel()
+	p.mu.Unlock()
+
+	p.wg.Wait()
+	return nil
+}
+
+// Dispatch checks if the given SNI refers to a connection for a Kubernetes
+// cluster through a Relay; if so, it returns true and the connection should be
+// ignored by the caller, otherwise the connection is left untouched and the
+// caller should handle it. The transcript contains data that was already sent
+// by the client (i.e. a TLS ClientHello) and will be sent to the agent before
+// forwarding more data from the connection. This method is intended to be used
+// with [relaytransport.SNIDispatchTransportCredentials].
+func (p *PassiveForwarder) Dispatch(serverName string, transcript *bytes.Buffer, rawConn net.Conn) (dispatched bool) {
+	sniPrefix, ok := strings.CutSuffix(serverName, SNISuffix)
+	if !ok {
+		return false
+	}
+
+	p.mu.Lock()
+	if p.ctx.Err() != nil {
+		p.mu.Unlock()
+		_ = rawConn.Close()
+		// the forwarder was closed and Close has potentially already returned,
+		// so we shouldn't spawn additional goroutines
+		return true
+	}
+	p.wg.Go(func() {
+		p.forward(sniPrefix, transcript, rawConn)
+	})
+	p.mu.Unlock()
+
+	return true
+}
+
+func (p *PassiveForwarder) forward(sniPrefix string, transcript *bytes.Buffer, clientConn net.Conn) {
+	ctx := p.ctx
+	log := p.log.With(
+		"client_addr", clientConn.RemoteAddr().String(),
+	)
+	// TODO(espadolini): there's no way to signal errors to the client, so the
+	// best we can do is to cut the connection abruptly - unfortunately, that
+	// will cause clients such as kubectl to treat requests as retriable, and
+	// it's only after a lengthy timeout that an error will actually be reported
+	// to the user. Then again, this same behavior also means that if the
+	// connection failure is caused by a temporary failure due to missing
+	// tunnels and the likes, the following retries have a chance of succeeding,
+	// so this behavior is not all bad.
+	defer clientConn.Close()
+
+	sniPrefix = strings.ToLower(sniPrefix)
+	encodedHash, ok := strings.CutPrefix(sniPrefix, SNIPrefixForKubeCluster)
+	if !ok || len(encodedHash) != encodedHashLen {
+		log.DebugContext(ctx,
+			"Received unsupported SNI for kube relay forwarding",
+			"sni_prefix", sniPrefix,
+		)
+		return
+	}
+	var desiredHash [hashLen]byte
+	if _, err := base32hex.Decode(desiredHash[:], []byte(encodedHash)); err != nil {
+		log.DebugContext(ctx,
+			"Received malformed hash in SNI for kube cluster relay forwarding",
+			"encoded_hash", encodedHash,
+		)
+		return
+	}
+
+	kubeClusterName, cached := p.resolvedNames.Get(desiredHash)
+	kubeServers, err := p.getKubeServersWithFilter(
+		ctx,
+		func(ks readonly.KubeServer) bool {
+			if ks.GetRelayGroup() != p.groupName {
+				return false
+			}
+
+			if kubeClusterName != "" {
+				return ks.GetCluster().GetName() == kubeClusterName
+			}
+
+			if desiredHash == hashForTarget(p.clusterName, ks.GetCluster().GetName()) {
+				kubeClusterName = ks.GetCluster().GetName()
+				if !cached {
+					p.resolvedNames.Add(desiredHash, kubeClusterName)
+				}
+				return true
+			}
+			return false
+		},
+	)
+	if err != nil {
+		if ctx.Err() != nil {
+			return
+		}
+		log.ErrorContext(ctx,
+			"Failed to retrieve kube servers for routing",
+			"encoded_hash", encodedHash,
+			"error", err,
+		)
+		return
+	}
+	if len(kubeServers) < 1 {
+		log.WarnContext(ctx,
+			"No kube server found for the desired hash in the local relay group",
+			"encoded_hash", encodedHash,
+		)
+		return
+	}
+
+	var agentConn net.Conn
+	var agentHostID string
+	for kubeServer := range healthcheck.OrderByTargetHealthStatus(kubeServers) {
+		target := kubeServer.GetHostID() + "." + p.clusterName
+
+		localTunnelConn, err := p.localDial(ctx,
+			target, apitypes.KubeTunnel,
+			clientConn.RemoteAddr(), clientConn.LocalAddr(),
+		)
+		if err == nil {
+			agentConn = localTunnelConn
+			agentHostID = kubeServer.GetHostID()
+			break
+		}
+
+		if !trace.IsNotFound(err) {
+			log.DebugContext(ctx,
+				"Failed to dial target in local relay tunnel server, trying peer dialing",
+				"agent_host_id", kubeServer.GetHostID(),
+				"error", err,
+			)
+		}
+
+		peerTunnelConn, err := p.peerDial(ctx,
+			target, apitypes.KubeTunnel,
+			slices.Clone(kubeServer.GetRelayIDs()),
+			clientConn.RemoteAddr(), clientConn.LocalAddr(),
+		)
+		if err == nil {
+			agentConn = peerTunnelConn
+			agentHostID = kubeServer.GetHostID()
+			break
+		}
+
+		log.DebugContext(ctx,
+			"Failed to dial target through relay peering, trying next target if any",
+			"agent_host_id", kubeServer.GetHostID(),
+			"error", err,
+		)
+	}
+	if agentConn == nil {
+		if ctx.Err() != nil {
+			return
+		}
+		log.WarnContext(ctx,
+			"Failed to open tunnel connection to any agent connected to this relay group serving the desired cluster",
+			"kube_cluster", kubeClusterName,
+		)
+		return
+	}
+	defer agentConn.Close()
+
+	defer context.AfterFunc(ctx, func() {
+		// this will unblock both sides of the copy (it's not sufficient to
+		// close just one, it might've already half-closed and the other
+		// direction might get stuck reading)
+		_ = agentConn.Close()
+		_ = clientConn.Close()
+	})()
+
+	log = log.With(
+		"kube_cluster", kubeClusterName,
+		"agent_host_id", agentHostID,
+	)
+	log.InfoContext(ctx,
+		"Forwarding kube connection to agent",
+	)
+	defer log.DebugContext(ctx,
+		"Done forwarding kube connection to agent",
+	)
+
+	var wg sync.WaitGroup
+	defer wg.Wait()
+
+	wg.Go(func() {
+		defer log.Log(ctx, logutils.TraceLevel,
+			"Done copying data from client to agent",
+		)
+		if _, err := io.Copy(agentConn, transcript); err != nil {
+			log.Log(ctx, traceIfOKNetworkError(err, slog.LevelDebug),
+				"Got an error copying data from client transcript to agent",
+				"error", err,
+			)
+			_ = clientConn.Close()
+			_ = agentConn.Close()
+			return
+		}
+		if _, err := io.Copy(agentConn, clientConn); err != nil {
+			log.Log(ctx, traceIfOKNetworkError(err, slog.LevelDebug),
+				"Got an error copying data from client to agent",
+				"error", err,
+			)
+			_ = clientConn.Close()
+			_ = agentConn.Close()
+			return
+		}
+	})
+
+	wg.Go(func() {
+		defer log.Log(ctx, logutils.TraceLevel,
+			"Done copying data from agent to client",
+		)
+		if _, err := io.Copy(clientConn, agentConn); err != nil {
+			log.Log(ctx, traceIfOKNetworkError(err, slog.LevelDebug),
+				"Got an error copying data from agent to client",
+				"error", err,
+			)
+			_ = clientConn.Close()
+			_ = agentConn.Close()
+			return
+		}
+	})
+}
+
+func traceIfOKNetworkError(err error, l slog.Level) slog.Level {
+	if err == nil || utils.IsOKNetworkError(err) {
+		return logutils.TraceLevel
+	}
+	return l
+}

lib/kube/relay/passive_forwarder_test.go

@@ -0,0 +1,145 @@
+// Teleport
+// Copyright (C) 2025 Gravitational, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package relay
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"net"
+	"os"
+	"testing"
+
+	"github.com/gravitational/trace"
+	"github.com/stretchr/testify/require"
+
+	apitypes "github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/services/readonly"
+	"github.com/gravitational/teleport/lib/utils/log/logtest"
+)
+
+func TestMain(m *testing.M) {
+	logtest.InitLogger(testing.Verbose)
+	os.Exit(m.Run())
+}
+
+func TestPassiveForwarder_Selection(t *testing.T) {
+	kubeServers := []*apitypes.KubernetesServerV3{
+		// not in any group
+		{
+			Spec: apitypes.KubernetesServerSpecV3{
+				RelayGroup: "",
+				HostID:     "host-a",
+				Cluster: &apitypes.KubernetesClusterV3{
+					Metadata: apitypes.Metadata{
+						Name: "cluster-a",
+					},
+				},
+			},
+		},
+		// in the group, unhealthy
+		{
+			Spec: apitypes.KubernetesServerSpecV3{
+				RelayGroup: "thisgroup",
+				HostID:     "host-b",
+				Cluster: &apitypes.KubernetesClusterV3{
+					Metadata: apitypes.Metadata{
+						Name: "cluster-b",
+					},
+				},
+			},
+			Status: &apitypes.KubernetesServerStatusV3{
+				TargetHealth: &apitypes.TargetHealth{
+					Status: "unhealthy",
+				},
+			},
+		},
+		// not in the correct group
+		{
+			Spec: apitypes.KubernetesServerSpecV3{
+				RelayGroup: "othergroup",
+				HostID:     "host-c",
+				Cluster: &apitypes.KubernetesClusterV3{
+					Metadata: apitypes.Metadata{
+						Name: "cluster-c",
+					},
+				},
+			},
+		},
+		// in the right group, healthy
+		{
+			Spec: apitypes.KubernetesServerSpecV3{
+				RelayGroup: "thisgroup",
+				HostID:     "host-b2",
+				Cluster: &apitypes.KubernetesClusterV3{
+					Metadata: apitypes.Metadata{
+						Name: "cluster-b",
+					},
+				},
+			},
+			Status: &apitypes.KubernetesServerStatusV3{
+				TargetHealth: &apitypes.TargetHealth{
+					Status: "healthy",
+				},
+			},
+		},
+	}
+
+	var localDialed []string
+	fwd, err := NewPassiveForwarder(PassiveForwarderConfig{
+		Log:         slog.Default(),
+		ClusterName: "clustername",
+		GroupName:   "thisgroup",
+		GetKubeServersWithFilter: func(ctx context.Context, filter func(readonly.KubeServer) bool) ([]apitypes.KubeServer, error) {
+			var out []apitypes.KubeServer
+			for _, s := range kubeServers {
+				if filter(s) {
+					out = append(out, s.Copy())
+				}
+			}
+			return out, nil
+		},
+		LocalDial: func(ctx context.Context, hostID string, tunnelType apitypes.TunnelType, src, dst net.Addr) (net.Conn, error) {
+			localDialed = append(localDialed, hostID)
+			return nil, trace.NotFound("no")
+		},
+		PeerDial: func(ctx context.Context, hostID string, tunnelType apitypes.TunnelType, relayIDs []string, src, dst net.Addr) (net.Conn, error) {
+			return nil, trace.ConnectionProblem(nil, "nope")
+		},
+	})
+	require.NoError(t, err)
+	defer fwd.Close()
+
+	fwd.forward(SNILabelForKubeCluster("clustername", "cluster-a"), new(bytes.Buffer), (*noopConn)(nil))
+	require.Empty(t, localDialed)
+	fwd.forward(SNILabelForKubeCluster("clustername", "cluster-c"), new(bytes.Buffer), (*noopConn)(nil))
+	require.Empty(t, localDialed)
+	// host-b2 is healthy so it will always be tried before the unhealthy host-b
+	for range 100 {
+		localDialed = []string{}
+		fwd.forward(SNILabelForKubeCluster("clustername", "cluster-b"), new(bytes.Buffer), (*noopConn)(nil))
+		require.Equal(t, []string{"host-b2.clustername", "host-b.clustername"}, localDialed)
+	}
+}
+
+type noopConn struct {
+	net.Conn
+}
+
+func (*noopConn) Close() error         { return nil }
+func (*noopConn) LocalAddr() net.Addr  { return &net.TCPAddr{} }
+func (*noopConn) RemoteAddr() net.Addr { return &net.TCPAddr{} }

lib/relaytransport/sni.go

@@ -0,0 +1,158 @@
+// Teleport
+// Copyright (C) 2025 Gravitational, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package relaytransport
+
+import (
+	"bytes"
+	"crypto/tls"
+	"io"
+	"net"
+	"time"
+
+	"github.com/gravitational/trace"
+	"google.golang.org/grpc/credentials"
+)
+
+type SNIDispatchFunc = func(serverName string, transcript *bytes.Buffer, rawConn net.Conn) (dispatched bool)
+
+// SNIDispatchTransportCredentials is a wrapper around a
+// [credentials.TransportCredentials] that reads a TLS ClientHello from each
+// client connection and optionally dispatches the connection through a custom
+// function based on the Server Name Indication in the ClientHello, without
+// interacting with the connection otherwise.
+type SNIDispatchTransportCredentials struct {
+	_ struct{}
+
+	credentials.TransportCredentials
+
+	// DispatchFunc should return true if the connection with the given server
+	// name should not be handled by the gRPC server. The transcript buffer
+	// contains the data that was received from the connection (containing the
+	// TLS ClientHello and potentially more data).
+	DispatchFunc SNIDispatchFunc
+}
+
+var _ credentials.TransportCredentials = (*SNIDispatchTransportCredentials)(nil)
+
+// Clone implements [credentials.TransportCredentials].
+func (s *SNIDispatchTransportCredentials) Clone() credentials.TransportCredentials {
+	return &SNIDispatchTransportCredentials{
+		TransportCredentials: s.TransportCredentials.Clone(),
+		DispatchFunc:         s.DispatchFunc,
+	}
+}
+
+// ServerHandshake implements [credentials.TransportCredentials].
+func (s *SNIDispatchTransportCredentials) ServerHandshake(rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
+	// the deadline for rawConn was set by the grpc server machinery based on
+	// the server connection timeout, we should not apply additional timeouts
+
+	transcript := new(bytes.Buffer)
+	var serverName string
+	var receivedHello bool
+	if err := tls.Server(
+		&transcripterConn{conn: rawConn, transcript: transcript},
+		&tls.Config{
+			GetConfigForClient: func(chi *tls.ClientHelloInfo) (*tls.Config, error) {
+				serverName = chi.ServerName
+				receivedHello = true
+				return nil, io.EOF
+			},
+		},
+	).Handshake(); !receivedHello && err != nil {
+		_ = rawConn.Close()
+		return nil, nil, trace.Wrap(err)
+	}
+
+	if s.DispatchFunc(serverName, transcript, rawConn) {
+		return nil, nil, credentials.ErrConnDispatched
+	}
+
+	if transcript.Len() > 0 {
+		rawConn = &transcriptedConn{
+			Conn:       rawConn,
+			transcript: transcript,
+		}
+	}
+
+	return s.TransportCredentials.ServerHandshake(rawConn)
+}
+
+// transcripterConn is a net.Conn that also stores the data that was read from a
+// real net.Conn in a transcript, and for which every other operation is a noop.
+type transcripterConn struct {
+	transcript *bytes.Buffer
+	conn       net.Conn
+}
+
+var _ net.Conn = (*transcripterConn)(nil)
+
+// Close implements [net.Conn].
+func (p *transcripterConn) Close() error { return nil }
+
+// Read implements [net.Conn].
+func (p *transcripterConn) Read(b []byte) (n int, err error) {
+	n, err = p.conn.Read(b)
+	_, _ = p.transcript.Write(b[:n])
+	return n, err
+}
+
+// LocalAddr implements [net.Conn].
+func (p *transcripterConn) LocalAddr() net.Addr { return p.conn.LocalAddr() }
+
+// RemoteAddr implements [net.Conn].
+func (p *transcripterConn) RemoteAddr() net.Addr { return p.conn.RemoteAddr() }
+
+// SetDeadline implements [net.Conn].
+func (p *transcripterConn) SetDeadline(t time.Time) error { return nil }
+
+// SetReadDeadline implements [net.Conn].
+func (p *transcripterConn) SetReadDeadline(t time.Time) error { return nil }
+
+// SetWriteDeadline implements [net.Conn].
+func (p *transcripterConn) SetWriteDeadline(t time.Time) error { return nil }
+
+// Write implements [net.Conn].
+func (p *transcripterConn) Write(b []byte) (n int, err error) { return len(b), nil }
+
+// transcriptedConn is a net.Conn whose incoming data begins with the data in a
+// transcript.
+type transcriptedConn struct {
+	net.Conn
+	transcript *bytes.Buffer
+}
+
+var _ net.Conn = (*transcriptedConn)(nil)
+
+// Read implements [net.Conn].
+func (p *transcriptedConn) Read(b []byte) (n int, err error) {
+	if p.transcript == nil {
+		return p.Conn.Read(b)
+	}
+
+	// this is improper behavior since closing the Conn will not cause read
+	// errors until the transcript is exhausted and reads past the deadline will
+	// still succeed, but this net.Conn implementation is very much intended for
+	// a buffer containing a TLS ClientHello that gets fully read almost
+	// immediately anyway
+	n = copy(b, p.transcript.Next(len(b)))
+	if p.transcript.Len() < 1 {
+		p.transcript = nil
+	}
+
+	return n, nil
+}

lib/relaytransport/sni_test.go

@@ -0,0 +1,264 @@
+// Teleport
+// Copyright (C) 2025 Gravitational, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package relaytransport
+
+import (
+	"bytes"
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc/credentials"
+
+	"github.com/gravitational/teleport/lib/utils/uds"
+)
+
+func TestSNIDispatch_NoDispatch(t *testing.T) {
+	caCert := requireExampleServerCert(t)
+	caPool := x509.NewCertPool()
+	caPool.AddCert(caCert.Leaf)
+
+	var eg errgroup.Group
+	defer eg.Wait()
+
+	// net.Pipe is unbuffered and crypto/tls doesn't work well with unbuffered
+	// connections (even though it should)
+	sc, cc, err := uds.NewSocketpair(uds.SocketTypeStream)
+	require.NoError(t, err)
+	defer sc.Close()
+	defer cc.Close()
+
+	// from the point of view of a client, the SNI dispatcher does not exist
+	eg.Go(func() error {
+		defer cc.Close()
+		tc := tls.Client(cc, &tls.Config{
+			ServerName: "foo.example.com",
+			RootCAs:    caPool,
+		})
+		if err := tc.Handshake(); err != nil {
+			return err
+		}
+		if _, err := tc.Write([]byte("a")); err != nil {
+			return err
+		}
+		if _, err := io.ReadFull(tc, make([]byte, 1)); err != nil {
+			return err
+		}
+		return nil
+	})
+
+	// a connection going through the dispatcher with the wrong SNI will not be
+	// dispatched, and it will be passed to the inner TransportCredentials
+	var transportCredentialsCalled, dispatchFuncCalled int
+	var dispatchErr error
+	returnedConn, _, err := (&SNIDispatchTransportCredentials{
+		DispatchFunc: func(serverName string, transcript *bytes.Buffer, rawConn net.Conn) (dispatched bool) {
+			dispatchFuncCalled++
+			if serverName != "foo.example.com" {
+				dispatchErr = fmt.Errorf("expected foo.example.com, got SNI %+q", serverName)
+			}
+			return false
+		},
+		TransportCredentials: transportCredentialsFunc(func(rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
+			transportCredentialsCalled++
+			return rawConn, nil, nil
+		}),
+	}).ServerHandshake(sc)
+	require.NoError(t, err)
+	require.NoError(t, dispatchErr)
+	require.Equal(t, 1, dispatchFuncCalled)
+	require.Equal(t, 1, transportCredentialsCalled)
+
+	// prove that we can use the non-dispatched conn as is
+	tc := tls.Server(returnedConn, &tls.Config{
+		Certificates: []tls.Certificate{caCert},
+	})
+	require.NoError(t, tc.Handshake())
+
+	_, err = tc.Write([]byte("a"))
+	require.NoError(t, err)
+	_, err = io.ReadFull(tc, make([]byte, 1))
+	require.NoError(t, err)
+
+	// the client is happy
+	require.NoError(t, eg.Wait())
+}
+
+func TestSNIDispatch_YesDispatch(t *testing.T) {
+	caCert := requireExampleServerCert(t)
+	caPool := x509.NewCertPool()
+	caPool.AddCert(caCert.Leaf)
+
+	var eg errgroup.Group
+	defer eg.Wait()
+
+	// net.Pipe is unbuffered and crypto/tls doesn't work well with unbuffered
+	// connections (even though it should)
+	sc, cc, err := uds.NewSocketpair(uds.SocketTypeStream)
+	require.NoError(t, err)
+	defer sc.Close()
+	defer cc.Close()
+
+	// from the point of view of a client, the SNI dispatcher does not exist
+	eg.Go(func() error {
+		defer cc.Close()
+		tc := tls.Client(cc, &tls.Config{
+			ServerName: "bar.example.com",
+			RootCAs:    caPool,
+		})
+		if err := tc.Handshake(); err != nil {
+			return err
+		}
+		if _, err := tc.Write([]byte("a")); err != nil {
+			return err
+		}
+		if _, err := io.ReadFull(tc, make([]byte, 1)); err != nil {
+			return err
+		}
+		return nil
+	})
+
+	// a connection going through the dispatcher with the correct SNI will be
+	// dispatched and the inner TransportCredentials doesn't see it
+	var transportCredentialsCalled, dispatchFuncCalled int
+	var dispatchErr error
+	var dispatchedTranscript *bytes.Buffer
+	var dispatchedConn net.Conn
+	_, _, err = (&SNIDispatchTransportCredentials{
+		DispatchFunc: func(serverName string, transcript *bytes.Buffer, rawConn net.Conn) (dispatched bool) {
+			dispatchFuncCalled++
+			if serverName != "bar.example.com" {
+				dispatchErr = fmt.Errorf("expected bar.example.com, got SNI %+q", serverName)
+				return false
+			}
+			dispatchedTranscript = transcript
+			dispatchedConn = rawConn
+			return true
+		},
+		TransportCredentials: transportCredentialsFunc(func(rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
+			transportCredentialsCalled++
+			return rawConn, nil, nil
+		}),
+	}).ServerHandshake(sc)
+	require.ErrorIs(t, err, credentials.ErrConnDispatched)
+	require.NoError(t, dispatchErr)
+	require.Equal(t, 1, dispatchFuncCalled)
+	require.Equal(t, 0, transportCredentialsCalled)
+
+	// the DispatchFunc gets the actual connection object plus a transcript
+	require.Same(t, sc, dispatchedConn)
+
+	// a TLS ClientHello is in there
+	require.NotNil(t, dispatchedTranscript)
+	require.NotZero(t, dispatchedTranscript.Len())
+
+	type nestedConn struct {
+		net.Conn
+	}
+	type bypassReader struct {
+		io.Reader
+		nestedConn
+	}
+	// methods in the io.Reader will take priority over nestedConn
+	scWithTranscript := bypassReader{
+		Reader:     io.MultiReader(dispatchedTranscript, dispatchedConn),
+		nestedConn: nestedConn{dispatchedConn},
+	}
+
+	// prove that we can use the dispatched conn if it's preceded by the
+	// returned transcript
+	tc := tls.Server(scWithTranscript, &tls.Config{
+		Certificates: []tls.Certificate{caCert},
+	})
+	require.NoError(t, tc.Handshake())
+
+	_, err = tc.Write([]byte("a"))
+	require.NoError(t, err)
+	_, err = io.ReadFull(tc, make([]byte, 1))
+	require.NoError(t, err)
+
+	// the client is happy
+	require.NoError(t, eg.Wait())
+}
+
+type transportCredentialsFunc func(rawConn net.Conn) (net.Conn, credentials.AuthInfo, error)
+
+var _ credentials.TransportCredentials = transportCredentialsFunc(nil)
+
+// ServerHandshake implements [credentials.TransportCredentials].
+func (fn transportCredentialsFunc) ServerHandshake(c net.Conn) (net.Conn, credentials.AuthInfo, error) {
+	return fn(c)
+}
+
+// ClientHandshake implements [credentials.TransportCredentials].
+func (transportCredentialsFunc) ClientHandshake(context.Context, string, net.Conn) (net.Conn, credentials.AuthInfo, error) {
+	panic("unimplemented")
+}
+
+// Clone implements [credentials.TransportCredentials].
+func (transportCredentialsFunc) Clone() credentials.TransportCredentials {
+	panic("unimplemented")
+}
+
+// Info implements [credentials.TransportCredentials].
+func (transportCredentialsFunc) Info() credentials.ProtocolInfo {
+	panic("unimplemented")
+}
+
+// OverrideServerName implements [credentials.TransportCredentials].
+func (transportCredentialsFunc) OverrideServerName(string) error {
+	panic("unimplemented")
+}
+
+func requireExampleServerCert(t *testing.T) tls.Certificate {
+	t.Helper()
+
+	_, key, err := ed25519.GenerateKey(nil)
+	require.NoError(t, err)
+
+	cert := &x509.Certificate{
+		NotBefore:   time.Now().Add(-24 * time.Hour),
+		NotAfter:    time.Now().Add(24 * time.Hour),
+		KeyUsage:    x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
+		ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+
+		BasicConstraintsValid: true,
+		IsCA:                  true,
+
+		DNSNames: []string{"*.example.com"},
+	}
+	der, err := x509.CreateCertificate(rand.Reader, cert, cert, key.Public(), key)
+	require.NoError(t, err)
+
+	cert, err = x509.ParseCertificate(der)
+	require.NoError(t, err)
+
+	return tls.Certificate{
+		Certificate: [][]byte{der},
+		PrivateKey:  key,
+		Leaf:        cert,
+	}
+}

lib/relaytunnel/discover.go

@@ -43,6 +43,7 @@ type StaticDiscoverServiceServer struct {
 
 	RelayGroup            string
 	TargetConnectionCount int32
+	SupportedTunnelTypes  []string
 }
 
 var _ relaytunnelv1alpha.DiscoveryServiceServer = (*StaticDiscoverServiceServer)(nil)
@@ -52,6 +53,7 @@ func (d *StaticDiscoverServiceServer) Discover(ctx context.Context, req *relaytu
 	return &relaytunnelv1alpha.DiscoverResponse{
 		RelayGroup:            d.RelayGroup,
 		TargetConnectionCount: d.TargetConnectionCount,
+		SupportedTunnelTypes:  d.SupportedTunnelTypes,
 	}, nil
 }
 

lib/relaytunnel/tunnel_client.go

@@ -244,7 +244,29 @@ func (c *Client) discoverOnce(ctx context.Context) error {
 		return trace.Wrap(err)
 	}
 
-	c.updateDiscovery(ctx, resp.GetRelayGroup(), resp.GetTargetConnectionCount())
+	effectiveConnectionCount := resp.GetTargetConnectionCount()
+
+	supportedTunnelTypes := resp.GetSupportedTunnelTypes()
+	if len(supportedTunnelTypes) < 1 {
+		supportedTunnelTypes = []string{string(types.NodeTunnel)}
+	}
+
+	if !slices.Contains(supportedTunnelTypes, string(c.tunnelType)) {
+		// the relay group is not advertising support for the tunnel type we
+		// want, so we just stay at a connection count of 0 to not attempt
+		// opening connections that would not work anyway; the next check will
+		// try again, so if the relay is upgraded and starts supporting our
+		// tunnel type we will begin opening connections
+		c.log.ErrorContext(ctx,
+			"The specified relay server does not currently support the required tunnel type and no tunnels will be opened, will check again later",
+			"relay_addr", c.relayAddr,
+			"tunnel_type", string(c.tunnelType),
+			"supported_tunnel_types", supportedTunnelTypes,
+		)
+		effectiveConnectionCount = 0
+	}
+
+	c.updateDiscovery(ctx, resp.GetRelayGroup(), effectiveConnectionCount)
 
 	return nil
 }

lib/relaytunnel/tunnel_server.go

@@ -133,18 +133,18 @@ func (*grpcServerCredentials) ClientHandshake(ctx context.Context, authority str
 	return nil, nil, trace.NotImplemented("these transport credentials can only be used as a server")
 }
 
-// OverrideServerName implements implements [credentials.TransportCredentials].
+// OverrideServerName implements [credentials.TransportCredentials].
 func (*grpcServerCredentials) OverrideServerName(string) error {
 	return nil
 }
 
-// Clone implements implements [credentials.TransportCredentials].
+// Clone implements [credentials.TransportCredentials].
 func (s *grpcServerCredentials) Clone() credentials.TransportCredentials {
 	// s is immutable so there's no need to copy anything
 	return s
 }
 
-// Info implements implements [credentials.TransportCredentials].
+// Info implements [credentials.TransportCredentials].
 func (s *grpcServerCredentials) Info() credentials.ProtocolInfo {
 	return credentials.ProtocolInfo{
 		SecurityProtocol: "tls",
@@ -152,7 +152,7 @@ func (s *grpcServerCredentials) Info() credentials.ProtocolInfo {
 	}
 }
 
-// ServerHandshake implements implements [credentials.TransportCredentials].
+// ServerHandshake implements [credentials.TransportCredentials].
 func (s *grpcServerCredentials) ServerHandshake(rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
@@ -332,6 +332,8 @@ func (s *Server) handleYamuxTunnel(c io.ReadWriteCloser, clientID *tlsca.Identit
 		switch tunnelType {
 		case types.NodeTunnel:
 			requiredRole = types.RoleNode
+		case types.KubeTunnel:
+			requiredRole = types.RoleKube
 		default:
 			return trace.BadParameter("unsupported tunnel type %q", tunnelType)
 		}

lib/service/kubernetes.go

@@ -36,6 +36,7 @@ import (
 	"github.com/gravitational/teleport/lib/labels"
 	"github.com/gravitational/teleport/lib/modules"
 	"github.com/gravitational/teleport/lib/multiplexer"
+	"github.com/gravitational/teleport/lib/relaytunnel"
 	"github.com/gravitational/teleport/lib/reversetunnel"
 	"github.com/gravitational/teleport/lib/reversetunnelclient"
 	"github.com/gravitational/teleport/lib/services"
@@ -82,6 +83,7 @@ func (process *TeleportProcess) initKubernetesService(logger *slog.Logger, conn
 
 	teleportClusterName := conn.ClusterName()
 	proxyGetter := reversetunnel.NewConnectedProxyGetter()
+	relayInfoHolder := new(relaytunnel.InfoHolder)
 
 	// This service can run in 2 modes:
 	// 1. Reachable (by the proxy) - registers with auth server directly and
@@ -92,6 +94,7 @@ func (process *TeleportProcess) initKubernetesService(logger *slog.Logger, conn
 	// The listener exposes incoming connections over either mode.
 	var listener net.Listener
 	var agentPool *reversetunnel.AgentPool
+	var relayTunnelClient *relaytunnel.Client
 	switch {
 	// Filter out cases where both listen_addr and tunnel are set or both are
 	// not set.
@@ -152,6 +155,34 @@ func (process *TeleportProcess) initKubernetesService(logger *slog.Logger, conn
 			}
 		}()
 		logger.InfoContext(process.ExitContext(), "Started reverse tunnel client.")
+
+		if cfg.RelayServer != "" {
+			relayTunnelClient, err = relaytunnel.NewClient(relaytunnel.ClientConfig{
+				Log: logger,
+
+				GetCertificate: conn.ClientGetCertificate,
+				GetPool:        conn.ClientGetPool,
+				Ciphersuites:   cfg.CipherSuites,
+
+				TunnelType: types.KubeTunnel,
+				RelayAddr:  cfg.RelayServer,
+
+				HandleConnection: shtl.HandleConnection,
+				RelayInfoSetter:  relayInfoHolder.SetRelayInfo,
+			})
+			if err != nil {
+				return trace.Wrap(err)
+			}
+			if err := relayTunnelClient.Start(); err != nil {
+				return trace.Wrap(err)
+			}
+			defer func() {
+				if retErr != nil {
+					relayTunnelClient.Close()
+				}
+			}()
+			logger.InfoContext(process.ExitContext(), "Started relay tunnel client.")
+		}
 	}
 
 	var dynLabels *labels.Dynamic
@@ -253,6 +284,7 @@ func (process *TeleportProcess) initKubernetesService(logger *slog.Logger, conn
 		OnHeartbeat:          process.OnHeartbeat(teleport.ComponentKube),
 		GetRotation:          process.GetRotation,
 		ConnectedProxyGetter: proxyGetter,
+		RelayInfoGetter:      relayInfoHolder.GetRelayInfo,
 		ResourceMatchers:     cfg.Kube.ResourceMatchers,
 		StaticLabels:         cfg.Kube.StaticLabels,
 		DynamicLabels:        dynLabels,
@@ -300,6 +332,9 @@ func (process *TeleportProcess) initKubernetesService(logger *slog.Logger, conn
 			warnOnErr(process.ExitContext(), kubeServer.Close(), logger)
 			agentPool.Stop()
 		}
+		if relayTunnelClient != nil {
+			relayTunnelClient.Close()
+		}
 		if asyncEmitter != nil {
 			warnOnErr(process.ExitContext(), asyncEmitter.Close(), logger)
 		}

lib/service/relay.go

@@ -46,9 +46,11 @@ import (
 	"github.com/gravitational/teleport/lib/auth"
 	"github.com/gravitational/teleport/lib/auth/authclient"
 	"github.com/gravitational/teleport/lib/authz"
+	kuberelay "github.com/gravitational/teleport/lib/kube/relay"
 	multiplexergrpc "github.com/gravitational/teleport/lib/multiplexer/grpc"
 	"github.com/gravitational/teleport/lib/proxy"
 	"github.com/gravitational/teleport/lib/relaypeer"
+	"github.com/gravitational/teleport/lib/relaytransport"
 	"github.com/gravitational/teleport/lib/relaytunnel"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/srv"
@@ -141,6 +143,20 @@ func (process *TeleportProcess) runRelayService() error {
 	}
 	defer nodeWatcher.Close()
 
+	kubeServerWatcher, err := services.NewKubeServerWatcher(process.ExitContext(), services.KubeServerWatcherConfig{
+		ResourceWatcherConfig: services.ResourceWatcherConfig{
+			Component:    teleport.ComponentRelay,
+			Logger:       sublogger("kube_server_watcher"),
+			Client:       accessPoint,
+			MaxStaleness: time.Minute,
+		},
+		KubernetesServerGetter: accessPoint,
+	})
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	defer kubeServerWatcher.Close()
+
 	tunnelServer, err := relaytunnel.NewServer(relaytunnel.ServerConfig{
 		Log: sublogger("tunnel_server"),
 		GetCertificate: func(ctx context.Context) (*tls.Certificate, error) {
@@ -179,6 +195,10 @@ func (process *TeleportProcess) runRelayService() error {
 	relaytunnelv1alpha.RegisterDiscoveryServiceServer(tunnelGRPCServer, &relaytunnel.StaticDiscoverServiceServer{
 		RelayGroup:            process.Config.Relay.RelayGroup,
 		TargetConnectionCount: process.Config.Relay.TargetConnectionCount,
+		SupportedTunnelTypes: []string{
+			string(apitypes.NodeTunnel),
+			string(apitypes.KubeTunnel),
+		},
 	})
 
 	peerServer, err := relaypeer.NewServer(relaypeer.ServerConfig{
@@ -221,6 +241,35 @@ func (process *TeleportProcess) runRelayService() error {
 		return c, nil
 	}
 
+	relayPeerClient, err := relaypeer.NewClient(relaypeer.ClientConfig{
+		HostID:      conn.hostID,
+		ClusterName: conn.clusterName,
+		GroupName:   process.Config.Relay.RelayGroup,
+
+		AccessPoint: accessPoint,
+		Log:         sublogger("peer_client"),
+
+		GetCertificate: conn.ClientGetCertificate,
+		GetPool:        conn.ClientGetPool,
+		Ciphersuites:   process.Config.CipherSuites,
+	})
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
+	kubeForwarder, err := kuberelay.NewPassiveForwarder(kuberelay.PassiveForwarderConfig{
+		Log:                      sublogger("kube_forwarder"),
+		ClusterName:              conn.clusterName,
+		GroupName:                process.Config.Relay.RelayGroup,
+		GetKubeServersWithFilter: kubeServerWatcher.CurrentResourcesWithFilter,
+		LocalDial:                tunnelServer.Dial,
+		PeerDial:                 relayPeerClient.Dial,
+	})
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	defer kubeForwarder.Close()
+
 	var transportCreds credentials.TransportCredentials
 	{
 		tc, err := auth.NewTransportCredentials(auth.TransportCredentialsConfig{
@@ -236,26 +285,14 @@ func (process *TeleportProcess) runRelayService() error {
 		}
 		transportCreds = tc
 	}
+	transportCreds = &relaytransport.SNIDispatchTransportCredentials{
+		TransportCredentials: transportCreds,
+		DispatchFunc:         kubeForwarder.Dispatch,
+	}
 	if process.Config.Relay.TransportPROXYProtocol {
 		transportCreds = multiplexergrpc.PPV2ServerCredentials{TransportCredentials: transportCreds}
 	}
 
-	relayPeerClient, err := relaypeer.NewClient(relaypeer.ClientConfig{
-		HostID:      conn.hostID,
-		ClusterName: conn.clusterName,
-		GroupName:   process.Config.Relay.RelayGroup,
-
-		AccessPoint: accessPoint,
-		Log:         sublogger("relay_router"),
-
-		GetCertificate: conn.ClientGetCertificate,
-		GetPool:        conn.ClientGetPool,
-		Ciphersuites:   process.Config.CipherSuites,
-	})
-	if err != nil {
-		return trace.Wrap(err)
-	}
-
 	relayRouter, err := proxy.NewRelayRouter(proxy.RelayRouterConfig{
 		ClusterName: conn.clusterName,
 		GroupName:   process.Config.Relay.RelayGroup,

lib/service/service.go

@@ -141,6 +141,7 @@ import (
 	"github.com/gravitational/teleport/lib/join/legacyjoin"
 	kubegrpc "github.com/gravitational/teleport/lib/kube/grpc"
 	kubeproxy "github.com/gravitational/teleport/lib/kube/proxy"
+	kuberelay "github.com/gravitational/teleport/lib/kube/relay"
 	"github.com/gravitational/teleport/lib/labels"
 	"github.com/gravitational/teleport/lib/limiter"
 	"github.com/gravitational/teleport/lib/modules"
@@ -4336,6 +4337,9 @@ func (process *TeleportProcess) getAdditionalPrincipals(role types.SystemRole, h
 			utils.NetAddr{Addr: reversetunnelclient.LocalKubernetes},
 		)
 		addrs = append(addrs, process.Config.Kube.PublicAddrs...)
+		if process.Config.RelayServer != "" {
+			dnsNames = append(dnsNames, "*"+kuberelay.SNISuffix)
+		}
 	case types.RoleApp, types.RoleOkta:
 		principals = append(principals, hostUUID)
 	case types.RoleWindowsDesktop:

lib/services/readonly/readonly.go

@@ -272,6 +272,9 @@ type KubeServer interface {
 	GetCluster() types.KubeCluster
 	// GetProxyIDs returns a list of proxy ids this service is connected to.
 	GetProxyIDs() []string
+	// GetRelayGroup returns the name of the Relay group that the kube server is
+	// connected to.
+	GetRelayGroup() string
 }
 
 var _ KubeServer = types.KubeServer(nil)

proto/teleport/relaytunnel/v1alpha/discovery_service.proto

@@ -35,4 +35,8 @@ message DiscoverResponse {
   string relay_group = 1;
 
   int32 target_connection_count = 2;
+
+  // the list of api/types.TunnelTypes supported by this Relay group; if empty,
+  // it should be treated as containing only types.NodeTunnel ("node")
+  repeated string supported_tunnel_types = 3;
 }