From ded76989099c5d2982d4845570f072dd1c826dcc Mon Sep 17 00:00:00 2001 From: Edoardo Spadolini Date: Fri, 14 Nov 2025 23:25:00 +0100 Subject: [PATCH] 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 --- .../v1alpha/discovery_service.pb.go | 19 +- lib/auth/authclient/api.go | 2 + lib/cache/cache.go | 1 + lib/kube/proxy/server.go | 25 +- lib/kube/relay/hash.go | 54 +++ lib/kube/relay/hash_test.go | 69 +++ lib/kube/relay/passive_forwarder.go | 407 ++++++++++++++++++ lib/kube/relay/passive_forwarder_test.go | 145 +++++++ lib/relaytransport/sni.go | 158 +++++++ lib/relaytransport/sni_test.go | 264 ++++++++++++ lib/relaytunnel/discover.go | 2 + lib/relaytunnel/tunnel_client.go | 24 +- lib/relaytunnel/tunnel_server.go | 10 +- lib/service/kubernetes.go | 35 ++ lib/service/relay.go | 69 ++- lib/service/service.go | 4 + lib/services/readonly/readonly.go | 3 + .../v1alpha/discovery_service.proto | 4 + 18 files changed, 1264 insertions(+), 31 deletions(-) create mode 100644 lib/kube/relay/hash.go create mode 100644 lib/kube/relay/hash_test.go create mode 100644 lib/kube/relay/passive_forwarder.go create mode 100644 lib/kube/relay/passive_forwarder_test.go create mode 100644 lib/relaytransport/sni.go create mode 100644 lib/relaytransport/sni_test.go diff --git a/gen/proto/go/teleport/relaytunnel/v1alpha/discovery_service.pb.go b/gen/proto/go/teleport/relaytunnel/v1alpha/discovery_service.pb.go index 63bc1ff9c40..05c219af1e5 100644 --- a/gen/proto/go/teleport/relaytunnel/v1alpha/discovery_service.pb.go +++ b/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" diff --git a/lib/auth/authclient/api.go b/lib/auth/authclient/api.go index dcdfb64ff63..9403ab359b8 100644 --- a/lib/auth/authclient/api.go +++ b/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 diff --git a/lib/cache/cache.go b/lib/cache/cache.go index 9dfbce8eed3..585ebe8c05f 100644 --- a/lib/cache/cache.go +++ b/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}, diff --git a/lib/kube/proxy/server.go b/lib/kube/proxy/server.go index 2b0647fbff7..caf4c5eba64 100644 --- a/lib/kube/proxy/server.go +++ b/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 { diff --git a/lib/kube/relay/hash.go b/lib/kube/relay/hash.go new file mode 100644 index 00000000000..b7d2e13df4e --- /dev/null +++ b/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 . + +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[:]) +} diff --git a/lib/kube/relay/hash_test.go b/lib/kube/relay/hash_test.go new file mode 100644 index 00000000000..cd101e380a2 --- /dev/null +++ b/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 . + +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)) + } +} diff --git a/lib/kube/relay/passive_forwarder.go b/lib/kube/relay/passive_forwarder.go new file mode 100644 index 00000000000..cabe1b379d3 --- /dev/null +++ b/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 . + +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 +} diff --git a/lib/kube/relay/passive_forwarder_test.go b/lib/kube/relay/passive_forwarder_test.go new file mode 100644 index 00000000000..013d65cc9a8 --- /dev/null +++ b/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 . + +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{} } diff --git a/lib/relaytransport/sni.go b/lib/relaytransport/sni.go new file mode 100644 index 00000000000..f7d1ecdf298 --- /dev/null +++ b/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 . + +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 +} diff --git a/lib/relaytransport/sni_test.go b/lib/relaytransport/sni_test.go new file mode 100644 index 00000000000..199158e5bbc --- /dev/null +++ b/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 . + +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, + } +} diff --git a/lib/relaytunnel/discover.go b/lib/relaytunnel/discover.go index f8c16ccdd9b..5cab12b6c4e 100644 --- a/lib/relaytunnel/discover.go +++ b/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 } diff --git a/lib/relaytunnel/tunnel_client.go b/lib/relaytunnel/tunnel_client.go index c8e1bdd87b8..16f7679dcc1 100644 --- a/lib/relaytunnel/tunnel_client.go +++ b/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 } diff --git a/lib/relaytunnel/tunnel_server.go b/lib/relaytunnel/tunnel_server.go index b0914eefc8d..27b5b65dfdc 100644 --- a/lib/relaytunnel/tunnel_server.go +++ b/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) } diff --git a/lib/service/kubernetes.go b/lib/service/kubernetes.go index ad3e0f9f5b7..f3b51b95781 100644 --- a/lib/service/kubernetes.go +++ b/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) } diff --git a/lib/service/relay.go b/lib/service/relay.go index 483ae1cdd0a..d64a251c89f 100644 --- a/lib/service/relay.go +++ b/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, diff --git a/lib/service/service.go b/lib/service/service.go index e4e5b04f130..5e9b47024e3 100644 --- a/lib/service/service.go +++ b/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: diff --git a/lib/services/readonly/readonly.go b/lib/services/readonly/readonly.go index 8029448adda..657fc831496 100644 --- a/lib/services/readonly/readonly.go +++ b/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) diff --git a/proto/teleport/relaytunnel/v1alpha/discovery_service.proto b/proto/teleport/relaytunnel/v1alpha/discovery_service.proto index a16fcd67787..84a23396a4c 100644 --- a/proto/teleport/relaytunnel/v1alpha/discovery_service.proto +++ b/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; }