opensource/teleport
3
0
Fork 0
mirror of https://github.com/gravitational/teleport.git synced 2026-09-24 16:17:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Preparation for the QUIC proxy peering implementation (#48836)

Browse Source 
* Move the QUIC proxy peering envvar check to lib/config

* Move lib/proxy/clusterdial to lib/peer/dial

* Move peer.clientConn to lib/proxy/peer/internal

* Require QUIC peering label to be set to "yes"

* Move peer TLS verification funcs to lib/proxy/peer/internal

* Move QUICServer to lib/proxy/peer/quic

* Avoid stuttering in lib/proxy/peer/quic

* Remove useless quic.ServerConfig.CipherSuites param

* Fix log style in proxy.peer.quic

* Move duplicatePeerMsg into the function scope

Co-authored-by: Alan Parra <alan.parra@goteleport.com>

---------

Co-authored-by: Alan Parra <alan.parra@goteleport.com>
This commit is contained in:
Edoardo Spadolini
2024-11-13 16:06:01 +00:00
committed by GitHub
co-authored by Alan Parra
parent 3a65e3ebca
commit 5c2bbfd6b7
17 changed files with 362 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/types/constants.go
+4 -4
View File
@@ -1066,10 +1066,10 @@ const (
// group they should attempt to be connected to.
ProxyGroupGenerationLabel = TeleportInternalLabelPrefix + "proxygroup-gen"
// ProxyPeerQUICLabel is the internal-user label for proxy heartbeats that's
// used to signal that the proxy supports receiving proxy peering
// connections over QUIC.
ProxyPeerQUICLabel = TeleportInternalLabelPrefix + "proxy-peer-quic"
// UnstableProxyPeerQUICLabel is the internal-use label for proxy heartbeats
// that's used to signal that the proxy supports receiving proxy peering
// connections over QUIC. The value should be "yes".
UnstableProxyPeerQUICLabel = TeleportInternalLabelPrefix + "proxy-peer-quic"
// OktaAppNameLabel is the individual app name label.
OktaAppNameLabel = TeleportInternalLabelPrefix + "okta-app-name"
lib/config/configuration.go
+5
View File
@@ -2730,6 +2730,11 @@ func Configure(clf *CommandLineFlags, cfg *servicecfg.Config, legacyAppFlags boo
cfg.DebugService.Enabled = false
}
// TODO(espadolini): allow this when the implementation is merged
if false && os.Getenv("TELEPORT_UNSTABLE_QUIC_PROXY_PEERING") == "yes" {
cfg.Proxy.QUICProxyPeering = true
}
return nil
}
lib/proxy/clusterdial/dial.go
-60
View File
@@ -1,60 +0,0 @@
/*
* Teleport
* Copyright (C) 2023 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 clusterdial
import (
"net"
"github.com/gravitational/trace"
"github.com/gravitational/teleport/lib/proxy/peer"
"github.com/gravitational/teleport/lib/reversetunnelclient"
)
// ClusterDialerFunc is a function that implements a peer.ClusterDialer.
type ClusterDialerFunc func(clusterName string, request peer.DialParams) (net.Conn, error)
// Dial dials makes a dial request to the given cluster.
func (f ClusterDialerFunc) Dial(clusterName string, request peer.DialParams) (net.Conn, error) {
return f(clusterName, request)
}
// NewClusterDialer implements proxy.ClusterDialer for a reverse tunnel server.
func NewClusterDialer(server reversetunnelclient.Server) ClusterDialerFunc {
return func(clusterName string, request peer.DialParams) (net.Conn, error) {
site, err := server.GetSite(clusterName)
if err != nil {
return nil, trace.Wrap(err)
}
dialParams := reversetunnelclient.DialParams{
ServerID: request.ServerID,
ConnType: request.ConnType,
From: request.From,
To: request.To,
FromPeerProxy: true,
}
conn, err := site.Dial(dialParams)
if err != nil {
return nil, trace.Wrap(err)
}
return conn, nil
}
}
lib/proxy/peer/client.go
+32 -57
View File
@@ -44,6 +44,7 @@ import (
streamutils "github.com/gravitational/teleport/api/utils/grpc/stream"
"github.com/gravitational/teleport/lib/auth/authclient"
"github.com/gravitational/teleport/lib/defaults"
"github.com/gravitational/teleport/lib/proxy/peer/internal"
"github.com/gravitational/teleport/lib/services"
"github.com/gravitational/teleport/lib/utils"
)
@@ -94,11 +95,11 @@ type ClientConfig struct {
}
// connShuffler shuffles the order of client connections.
type connShuffler func([]clientConn)
type connShuffler func([]internal.ClientConn)
// randomConnShuffler returns a conn shuffler that randomizes the order of connections.
func randomConnShuffler() connShuffler {
return func(conns []clientConn) {
return func(conns []internal.ClientConn) {
rand.Shuffle(len(conns), func(i, j int) {
conns[i], conns[j] = conns[j], conns[i]
})
@@ -107,7 +108,7 @@ func randomConnShuffler() connShuffler {
// noopConnShutffler returns a conn shuffler that keeps the original connection ordering.
func noopConnShuffler() connShuffler {
return func([]clientConn) {}
return func([]internal.ClientConn) {}
}
// checkAndSetDefaults checks and sets default values
@@ -163,32 +164,6 @@ func (c *ClientConfig) checkAndSetDefaults() error {
return nil
}
// clientConn manages client connections to a specific peer proxy (with a fixed
// host ID and address).
type clientConn interface {
// peerID returns the host ID of the peer proxy.
peerID() string
// peerAddr returns the address of the peer proxy.
peerAddr() string
// dial opens a connection of a given tunnel type to a node with the given
// ID through the peer proxy managed by the clientConn.
dial(
nodeID string,
src net.Addr,
dst net.Addr,
tunnelType types.TunnelType,
) (net.Conn, error)
// close closes all connections and releases any background resources
// immediately.
close() error
// shutdown waits until all connections are closed or the context is done,
// then acts like close.
shutdown(context.Context)
}
// grpcClientConn manages client connections to a specific peer proxy over gRPC.
type grpcClientConn struct {
cc *grpc.ClientConn
@@ -205,13 +180,13 @@ type grpcClientConn struct {
count int
}
var _ clientConn = (*grpcClientConn)(nil)
var _ internal.ClientConn = (*grpcClientConn)(nil)
// peerID implements [clientConn].
func (c *grpcClientConn) peerID() string { return c.id }
// PeerID implements [internal.ClientConn].
func (c *grpcClientConn) PeerID() string { return c.id }
// peerAddr implements [clientConn].
func (c *grpcClientConn) peerAddr() string { return c.addr }
// PeerAddr implements [internal.ClientConn].
func (c *grpcClientConn) PeerAddr() string { return c.addr }
// maybeAcquire returns a non-nil release func if the grpcClientConn is
// currently allowed to open connections; i.e., if it hasn't fully shut down.
@@ -234,8 +209,8 @@ func (c *grpcClientConn) maybeAcquire() (release func()) {
})
}
// shutdown implements [clientConn].
func (c *grpcClientConn) shutdown(ctx context.Context) {
// Shutdown implements [internal.ClientConn].
func (c *grpcClientConn) Shutdown(ctx context.Context) {
defer c.cc.Close()
c.mu.Lock()
@@ -255,13 +230,13 @@ func (c *grpcClientConn) shutdown(ctx context.Context) {
}
}
// close implements [clientConn].
func (c *grpcClientConn) close() error {
// Close implements [internal.ClientConn].
func (c *grpcClientConn) Close() error {
return c.cc.Close()
}
// dial implements [clientConn].
func (c *grpcClientConn) dial(
// Dial implements [internal.ClientConn].
func (c *grpcClientConn) Dial(
nodeID string,
src net.Addr,
dst net.Addr,
@@ -335,7 +310,7 @@ type Client struct {
cancel context.CancelFunc
config ClientConfig
conns map[string]clientConn
conns map[string]internal.ClientConn
metrics *clientMetrics
reporter *reporter
}
@@ -360,7 +335,7 @@ func NewClient(config ClientConfig) (*Client, error) {
config: config,
ctx: closeContext,
cancel: cancel,
conns: make(map[string]clientConn),
conns: make(map[string]internal.ClientConn),
metrics: metrics,
reporter: reporter,
}
@@ -453,7 +428,7 @@ func (c *Client) updateConnections(proxies []types.Server) error {
}
var toDelete []string
toKeep := make(map[string]clientConn)
toKeep := make(map[string]internal.ClientConn)
for id, conn := range c.conns {
proxy, ok := toDial[id]
@@ -464,7 +439,7 @@ func (c *Client) updateConnections(proxies []types.Server) error {
}
// peer address changed
if conn.peerAddr() != proxy.GetPeerAddr() {
if conn.PeerAddr() != proxy.GetPeerAddr() {
toDelete = append(toDelete, id)
continue
}
@@ -485,8 +460,8 @@ func (c *Client) updateConnections(proxies []types.Server) error {
}
// establish new connections
_, supportsQuic := proxy.GetLabel(types.ProxyPeerQUICLabel)
conn, err := c.connect(id, proxy.GetPeerAddr(), supportsQuic)
supportsQUIC, _ := proxy.GetLabel(types.UnstableProxyPeerQUICLabel)
conn, err := c.connect(id, proxy.GetPeerAddr(), supportsQUIC == "yes")
if err != nil {
c.metrics.reportTunnelError(errorProxyPeerTunnelDial)
c.config.Log.DebugContext(c.ctx, "error dialing peer proxy", "peer_id", id, "peer_addr", proxy.GetPeerAddr())
@@ -503,7 +478,7 @@ func (c *Client) updateConnections(proxies []types.Server) error {
for _, id := range toDelete {
if conn, ok := c.conns[id]; ok {
go conn.shutdown(c.ctx)
go conn.Shutdown(c.ctx)
}
}
c.conns = toKeep
@@ -556,9 +531,9 @@ func (c *Client) Shutdown(ctx context.Context) {
var wg sync.WaitGroup
for _, conn := range c.conns {
wg.Add(1)
go func(conn clientConn) {
go func(conn internal.ClientConn) {
defer wg.Done()
conn.shutdown(ctx)
conn.Shutdown(ctx)
}(conn)
}
wg.Wait()
@@ -572,7 +547,7 @@ func (c *Client) Stop() error {
var errs []error
for _, conn := range c.conns {
if err := conn.close(); err != nil {
if err := conn.Close(); err != nil {
errs = append(errs, err)
}
}
@@ -627,7 +602,7 @@ func (c *Client) dial(
var errs []error
for _, clientConn := range conns {
conn, err := clientConn.dial(nodeID, src, dst, tunnelType)
conn, err := clientConn.Dial(nodeID, src, dst, tunnelType)
if err != nil {
errs = append(errs, trace.Wrap(err))
continue
@@ -643,13 +618,13 @@ func (c *Client) dial(
// otherwise.
// The boolean returned in the second argument is intended for testing purposes,
// to indicates whether the connection was cached or newly established.
func (c *Client) getConnections(proxyIDs []string) ([]clientConn, bool, error) {
func (c *Client) getConnections(proxyIDs []string) ([]internal.ClientConn, bool, error) {
if len(proxyIDs) == 0 {
return nil, false, trace.BadParameter("failed to dial: no proxy ids given")
}
ids := make(map[string]struct{})
var conns []clientConn
var conns []internal.ClientConn
// look for existing matching connections.
c.RLock()
@@ -686,8 +661,8 @@ func (c *Client) getConnections(proxyIDs []string) ([]clientConn, bool, error) {
continue
}
_, supportsQuic := proxy.GetLabel(types.ProxyPeerQUICLabel)
conn, err := c.connect(id, proxy.GetPeerAddr(), supportsQuic)
supportsQUIC, _ := proxy.GetLabel(types.UnstableProxyPeerQUICLabel)
conn, err := c.connect(id, proxy.GetPeerAddr(), supportsQUIC == "yes")
if err != nil {
c.metrics.reportTunnelError(errorProxyPeerTunnelDirectDial)
c.config.Log.DebugContext(c.ctx, "error direct dialing peer proxy", "peer_id", id, "peer_addr", proxy.GetPeerAddr())
@@ -707,7 +682,7 @@ func (c *Client) getConnections(proxyIDs []string) ([]clientConn, bool, error) {
defer c.Unlock()
for _, conn := range conns {
c.conns[conn.peerID()] = conn
c.conns[conn.PeerID()] = conn
}
c.config.connShuffler(conns)
@@ -715,7 +690,7 @@ func (c *Client) getConnections(proxyIDs []string) ([]clientConn, bool, error) {
}
// connect dials a new connection to proxyAddr.
func (c *Client) connect(peerID string, peerAddr string, supportsQUIC bool) (clientConn, error) {
func (c *Client) connect(peerID string, peerAddr string, supportsQUIC bool) (internal.ClientConn, error) {
if supportsQUIC && c.config.QUICTransport != nil {
panic("QUIC proxy peering is not implemented")
}
lib/proxy/peer/client_test.go
+2 -1
View File
@@ -30,6 +30,7 @@ import (
"github.com/gravitational/teleport/api/client/proto"
clientapi "github.com/gravitational/teleport/api/client/proto"
"github.com/gravitational/teleport/api/types"
"github.com/gravitational/teleport/lib/proxy/peer/internal"
"github.com/gravitational/teleport/lib/utils"
)
@@ -208,7 +209,7 @@ func TestBackupClient(t *testing.T) {
require.True(t, dialCalled)
}
func waitForGRPCConns(t *testing.T, conns map[string]clientConn, d time.Duration) {
func waitForGRPCConns(t *testing.T, conns map[string]internal.ClientConn, d time.Duration) {
require.Eventually(t, func() bool {
for _, conn := range conns {
// panic if we hit a non-grpc client conn
lib/proxy/peer/credentials.go
+3 -4
View File
@@ -27,6 +27,7 @@ import (
"google.golang.org/grpc/credentials"
"github.com/gravitational/teleport/api/types"
"github.com/gravitational/teleport/lib/proxy/peer/internal"
"github.com/gravitational/teleport/lib/tlsca"
)
@@ -74,15 +75,13 @@ func (c *clientCredentials) ClientHandshake(ctx context.Context, laddr string, c
}
if err := validatePeer(c.peerID, identity); err != nil {
c.log.ErrorContext(ctx, duplicatePeerMsg, "peer_addr", c.peerAddr, "peer_id", c.peerID)
internal.LogDuplicatePeer(ctx, c.log, slog.LevelError, "peer_addr", c.peerAddr, "peer_id", c.peerID)
return nil, nil, trace.Wrap(err)
}
return conn, authInfo, nil
}
const duplicatePeerMsg = "Detected multiple Proxy Peers with the same public address when connecting to a Proxy which can lead to inconsistent state and problems establishing sessions. For best results ensure that `peer_public_addr` is unique per proxy and not a load balancer."
// getIdentity returns a [tlsca.Identity] that is created from the certificate
// presented during the TLS handshake.
func getIdentity(authInfo credentials.AuthInfo) (*tlsca.Identity, error) {
@@ -121,5 +120,5 @@ func validatePeer(peerID string, identity *tlsca.Identity) error {
return nil
}
return trace.AccessDenied("connected to unexpected proxy")
return trace.Wrap(internal.WrongProxyError{})
}
lib/proxy/peer/dial/dial.go
+37
View File
@@ -0,0 +1,37 @@
// Teleport
// Copyright (C) 2024 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 peerdial
import (
"net"
"github.com/gravitational/teleport/api/types"
"github.com/gravitational/teleport/lib/utils"
)
// Dialer dials a node in the given cluster.
type Dialer interface {
Dial(clusterName string, request DialParams) (net.Conn, error)
}
// DialParams defines the target for a [Dialer.Dial].
type DialParams struct {
From *utils.NetAddr
To *utils.NetAddr
ServerID string
ConnType types.TunnelType
}
lib/proxy/peer/helpers_test.go
+1 -1
View File
@@ -224,7 +224,7 @@ func setupServer(t *testing.T, name string, serverCA, clientCA *tlsca.CertAuthor
clientCAs.AddCert(clientCA.Cert)
config := ServerConfig{
ClusterDialer: &mockClusterDialer{},
Dialer: &mockClusterDialer{},
GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
return &tlsCert, nil
},
lib/proxy/peer/internal/clientconn.go
+50
View File
@@ -0,0 +1,50 @@
// Teleport
// Copyright (C) 2024 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 internal
import (
"context"
"net"
"github.com/gravitational/teleport/api/types"
)
// ClientConn manages client connections to a specific peer proxy (with a fixed
// host ID and address).
type ClientConn interface {
// PeerID returns the host ID of the peer proxy.
PeerID() string
// PeerAddr returns the address of the peer proxy.
PeerAddr() string
// Dial opens a connection of a given tunnel type to a node with the given
// ID through the peer proxy managed by the clientConn.
Dial(
nodeID string,
src net.Addr,
dst net.Addr,
tunnelType types.TunnelType,
) (net.Conn, error)
// Close closes all connections and releases any background resources
// immediately.
Close() error
// Shutdown waits until all connections are closed or the context is done,
// then acts like Close.
Shutdown(context.Context)
}
lib/proxy/peer/internal/tls.go
+98
View File
@@ -0,0 +1,98 @@
// Teleport
// Copyright (C) 2024 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 internal
import (
"context"
"crypto/x509"
"log/slog"
"slices"
"github.com/gravitational/trace"
"github.com/gravitational/teleport/api/types"
"github.com/gravitational/teleport/lib/tlsca"
)
// VerifyPeerCertificateIsProxy is a function usable as a
// [tls.Config.VerifyPeerCertificate] callback to enforce that the connected TLS
// client is using Proxy credentials.
func VerifyPeerCertificateIsProxy(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error {
if len(verifiedChains) < 1 {
return trace.AccessDenied("missing client certificate (this is a bug)")
}
clientCert := verifiedChains[0][0]
clientIdentity, err := tlsca.FromSubject(clientCert.Subject, clientCert.NotAfter)
if err != nil {
return trace.Wrap(err)
}
if !slices.Contains(clientIdentity.Groups, string(types.RoleProxy)) {
return trace.AccessDenied("expected Proxy client credentials")
}
return nil
}
// VerifyPeerCertificateIsSpecificProxy returns a function usable as a
// [tls.Config.VerifyPeerCertificate] callback to enforce that the connected TLS
// server is using Proxy credentials and has the expected host ID.
func VerifyPeerCertificateIsSpecificProxy(peerID string) func(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error {
return func(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error {
if len(verifiedChains) < 1 {
return trace.AccessDenied("missing server certificate (this is a bug)")
}
clientCert := verifiedChains[0][0]
clientIdentity, err := tlsca.FromSubject(clientCert.Subject, clientCert.NotAfter)
if err != nil {
return trace.Wrap(err)
}
if !slices.Contains(clientIdentity.Groups, string(types.RoleProxy)) {
return trace.AccessDenied("expected Proxy server credentials")
}
if clientIdentity.Username != peerID {
return trace.Wrap(WrongProxyError{})
}
return nil
}
}
// LogDuplicatePeer should be used to log a message if a proxy peering client
// connects to a Proxy that did not have the expected host ID.
func LogDuplicatePeer(ctx context.Context, log *slog.Logger, level slog.Level, args ...any) {
const duplicatePeerMsg = "" +
"Detected multiple Proxy Peers with the same public address when connecting to a Proxy which can lead to inconsistent state and problems establishing sessions. " +
"For best results ensure that `peer_public_addr` is unique per proxy and not a load balancer."
log.Log(ctx, level, duplicatePeerMsg, args...)
}
// WrongProxyError signals that a proxy peering client has connected to a Proxy
// that did not have the expected host ID.
type WrongProxyError struct{}
func (WrongProxyError) Error() string {
return "connected to unexpected proxy"
}
func (e WrongProxyError) Unwrap() error {
return &trace.AccessDeniedError{
Message: e.Error(),
}
}
lib/proxy/peer/quicserver.go → lib/proxy/peer/quic/server.go
+31 -39
View File
@@ -1,22 +1,20 @@
/*
* Teleport
* Copyright (C) 2023 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/>.
*/
// Teleport
// Copyright (C) 2024 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 peer
package quic
import (
"context"
@@ -28,22 +26,16 @@ import (
"github.com/quic-go/quic-go"
"github.com/gravitational/teleport"
peerdial "github.com/gravitational/teleport/lib/proxy/peer/dial"
)
// QUICServerConfig holds the parameters for [NewQUICServer].
type QUICServerConfig struct {
// ServerConfig holds the parameters for [NewServer].
type ServerConfig struct {
Log *slog.Logger
// ClusterDialer is the dialer used to open connections to agents on behalf
// Dialer is the dialer used to open connections to agents on behalf
// of the peer proxies. Required.
ClusterDialer ClusterDialer
Dialer peerdial.Dialer
// CipherSuites is the set of TLS ciphersuites to be used by the server.
//
// Note: it won't actually have an effect, since QUIC always uses (the DTLS
// equivalent of) TLS 1.3, and TLS 1.3 ciphersuites can't be configured in
// crypto/tls, but for consistency's sake this should be passed along from
// the agent configuration.
CipherSuites []uint16
// GetCertificate should return the server certificate at time of use. It
// should be a certificate with the Proxy host role. Required.
GetCertificate func(*tls.ClientHelloInfo) (*tls.Certificate, error)
@@ -54,7 +46,7 @@ type QUICServerConfig struct {
GetClientCAs func(*tls.ClientHelloInfo) (*x509.CertPool, error)
}
func (c *QUICServerConfig) checkAndSetDefaults() error {
func (c *ServerConfig) checkAndSetDefaults() error {
if c.Log == nil {
c.Log = slog.Default()
}
@@ -63,8 +55,8 @@ func (c *QUICServerConfig) checkAndSetDefaults() error {
teleport.Component(teleport.ComponentProxy, "qpeer"),
)
if c.ClusterDialer == nil {
return trace.BadParameter("missing cluster dialer")
if c.Dialer == nil {
return trace.BadParameter("missing Dialer")
}
if c.GetCertificate == nil {
@@ -77,11 +69,11 @@ func (c *QUICServerConfig) checkAndSetDefaults() error {
return nil
}
// QUICServer is a proxy peering server that uses the QUIC protocol.
type QUICServer struct{}
// Server is a proxy peering server that uses the QUIC protocol.
type Server struct{}
// NewQUICServer returns a [QUICServer] with the given config.
func NewQUICServer(cfg QUICServerConfig) (*QUICServer, error) {
// NewServer returns a [Server] with the given config.
func NewServer(cfg ServerConfig) (*Server, error) {
if err := cfg.checkAndSetDefaults(); err != nil {
return nil, trace.Wrap(err)
}
@@ -90,19 +82,19 @@ func NewQUICServer(cfg QUICServerConfig) (*QUICServer, error) {
// Serve opens a listener and serves incoming connection. Returns after calling
// Close or Shutdown.
func (s *QUICServer) Serve(t *quic.Transport) error {
func (s *Server) Serve(t *quic.Transport) error {
panic("QUIC proxy peering is not implemented")
}
// Close stops listening for incoming connections and ungracefully terminates
// all the existing ones.
func (s *QUICServer) Close() error {
func (s *Server) Close() error {
panic("QUIC proxy peering is not implemented")
}
// Shutdown stops listening for incoming connections and waits until the
// existing ones are closed or until the context expires. If the context
// expires, running connections are ungracefully terminated.
func (s *QUICServer) Shutdown(ctx context.Context) error {
func (s *Server) Shutdown(ctx context.Context) error {
panic("QUIC proxy peering is not implemented")
}
lib/proxy/peer/server.go
+15 -33
View File
@@ -25,7 +25,6 @@ import (
"log/slog"
"math"
"net"
"slices"
"time"
"github.com/gravitational/trace"
@@ -36,9 +35,9 @@ import (
"github.com/gravitational/teleport"
"github.com/gravitational/teleport/api/client/proto"
"github.com/gravitational/teleport/api/metadata"
"github.com/gravitational/teleport/api/types"
"github.com/gravitational/teleport/api/utils/grpc/interceptors"
"github.com/gravitational/teleport/lib/tlsca"
peerdial "github.com/gravitational/teleport/lib/proxy/peer/dial"
"github.com/gravitational/teleport/lib/proxy/peer/internal"
"github.com/gravitational/teleport/lib/utils"
)
@@ -49,8 +48,8 @@ const (
// ServerConfig configures a Server instance.
type ServerConfig struct {
Log *slog.Logger
ClusterDialer ClusterDialer
Log *slog.Logger
Dialer peerdial.Dialer
CipherSuites []uint16
GetCertificate func(*tls.ClientHelloInfo) (*tls.Certificate, error)
@@ -71,8 +70,8 @@ func (c *ServerConfig) checkAndSetDefaults() error {
teleport.Component(teleport.ComponentProxy, "peer"),
)
if c.ClusterDialer == nil {
return trace.BadParameter("missing cluster dialer server")
if c.Dialer == nil {
return trace.BadParameter("missing Dialer")
}
if c.GetCertificate == nil {
@@ -84,8 +83,8 @@ func (c *ServerConfig) checkAndSetDefaults() error {
if c.service == nil {
c.service = &proxyService{
c.ClusterDialer,
c.Log,
dialer: c.Dialer,
log: c.Log,
}
}
@@ -94,9 +93,9 @@ func (c *ServerConfig) checkAndSetDefaults() error {
// Server is a proxy service server using grpc and tls.
type Server struct {
log *slog.Logger
clusterDialer ClusterDialer
server *grpc.Server
log *slog.Logger
dialer peerdial.Dialer
server *grpc.Server
}
// NewServer creates a new proxy server instance.
@@ -117,7 +116,7 @@ func NewServer(cfg ServerConfig) (*Server, error) {
tlsConfig.NextProtos = []string{"h2"}
tlsConfig.GetCertificate = cfg.GetCertificate
tlsConfig.ClientAuth = tls.RequireAndVerifyClientCert
tlsConfig.VerifyPeerCertificate = verifyPeerCertificateIsProxy
tlsConfig.VerifyPeerCertificate = internal.VerifyPeerCertificateIsProxy
getClientCAs := cfg.GetClientCAs
tlsConfig.GetConfigForClient = func(chi *tls.ClientHelloInfo) (*tls.Config, error) {
@@ -158,29 +157,12 @@ func NewServer(cfg ServerConfig) (*Server, error) {
proto.RegisterProxyServiceServer(server, cfg.service)
return &Server{
log: cfg.Log,
clusterDialer: cfg.ClusterDialer,
server: server,
log: cfg.Log,
dialer: cfg.Dialer,
server: server,
}, nil
}
func verifyPeerCertificateIsProxy(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error {
if len(verifiedChains) < 1 {
return trace.AccessDenied("missing client certificate (this is a bug)")
}
clientCert := verifiedChains[0][0]
clientIdentity, err := tlsca.FromSubject(clientCert.Subject, clientCert.NotAfter)
if err != nil {
return trace.Wrap(err)
}
if !slices.Contains(clientIdentity.Groups, string(types.RoleProxy)) {
return trace.AccessDenied("expected Proxy client credentials")
}
return nil
}
// Serve starts the proxy server.
func (s *Server) Serve(l net.Listener) error {
if err := s.server.Serve(l); err != nil {
lib/proxy/peer/service.go
+4 -17
View File
@@ -20,21 +20,20 @@ package peer
import (
"log/slog"
"net"
"strings"
"github.com/gravitational/trace"
"github.com/gravitational/teleport/api/client/proto"
"github.com/gravitational/teleport/api/types"
streamutils "github.com/gravitational/teleport/api/utils/grpc/stream"
peerdial "github.com/gravitational/teleport/lib/proxy/peer/dial"
"github.com/gravitational/teleport/lib/utils"
)
// proxyService implements the grpc ProxyService.
type proxyService struct {
clusterDialer ClusterDialer
log *slog.Logger
dialer peerdial.Dialer
log *slog.Logger
}
// DialNode opens a bidirectional stream to the requested node.
@@ -75,7 +74,7 @@ func (s *proxyService) DialNode(stream proto.ProxyService_DialNodeServer) error
AddrNetwork: dial.Destination.Network,
}
nodeConn, err := s.clusterDialer.Dial(clusterName, DialParams{
nodeConn, err := s.dialer.Dial(clusterName, peerdial.DialParams{
From: source,
To: destination,
ServerID: dial.NodeID,
@@ -116,15 +115,3 @@ func splitServerID(address string) (string, string, error) {
return split[0], strings.Join(split[1:], "."), nil
}
// ClusterDialer dials a node in the given cluster.
type ClusterDialer interface {
Dial(clusterName string, request DialParams) (net.Conn, error)
}
type DialParams struct {
From *utils.NetAddr
To *utils.NetAddr
ServerID string
ConnType types.TunnelType
}
lib/proxy/peer/service_test.go
+5 -4
View File
@@ -31,13 +31,14 @@ import (
"github.com/gravitational/teleport/api/client/proto"
"github.com/gravitational/teleport/api/types"
peerdial "github.com/gravitational/teleport/lib/proxy/peer/dial"
)
type mockClusterDialer struct {
MockDialCluster func(string, DialParams) (net.Conn, error)
MockDialCluster func(string, peerdial.DialParams) (net.Conn, error)
}
func (m *mockClusterDialer) Dial(clusterName string, request DialParams) (net.Conn, error) {
func (m *mockClusterDialer) Dial(clusterName string, request peerdial.DialParams) (net.Conn, error) {
if m.MockDialCluster == nil {
return nil, trace.NotImplemented("")
}
@@ -93,8 +94,8 @@ func TestSendReceive(t *testing.T) {
}
local, remote := net.Pipe()
service.clusterDialer = &mockClusterDialer{
MockDialCluster: func(clusterName string, request DialParams) (net.Conn, error) {
service.dialer = &mockClusterDialer{
MockDialCluster: func(clusterName string, request peerdial.DialParams) (net.Conn, error) {
require.Equal(t, "test-cluster", clusterName)
require.Equal(t, dialRequest.TunnelType, request.ConnType)
require.Equal(t, dialRequest.NodeID, request.ServerID)
lib/reversetunnelclient/peer.go
+57
View File
@@ -0,0 +1,57 @@
// Teleport
// Copyright (C) 2024 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 reversetunnelclient
import (
"net"
"github.com/gravitational/trace"
peerdial "github.com/gravitational/teleport/lib/proxy/peer/dial"
)
// PeerDialerFunc is a function that implements [peerdial.Dialer].
type PeerDialerFunc func(clusterName string, request peerdial.DialParams) (net.Conn, error)
// Dial implements [peerdial.Dialer].
func (f PeerDialerFunc) Dial(clusterName string, request peerdial.DialParams) (net.Conn, error) {
return f(clusterName, request)
}
// NewPeerDialer implements [peerdial.Dialer] for a reverse tunnel server.
func NewPeerDialer(server Tunnel) PeerDialerFunc {
return func(clusterName string, request peerdial.DialParams) (net.Conn, error) {
site, err := server.GetSite(clusterName)
if err != nil {
return nil, trace.Wrap(err)
}
dialParams := DialParams{
ServerID: request.ServerID,
ConnType: request.ConnType,
From: request.From,
To: request.To,
FromPeerProxy: true,
}
conn, err := site.Dial(dialParams)
if err != nil {
return nil, trace.Wrap(err)
}
return conn, nil
}
}
lib/service/service.go
+13 -15
View File
@@ -137,8 +137,8 @@ import (
"github.com/gravitational/teleport/lib/openssh"
"github.com/gravitational/teleport/lib/plugin"
"github.com/gravitational/teleport/lib/proxy"
"github.com/gravitational/teleport/lib/proxy/clusterdial"
"github.com/gravitational/teleport/lib/proxy/peer"
peerquic "github.com/gravitational/teleport/lib/proxy/peer/quic"
"github.com/gravitational/teleport/lib/resumption"
"github.com/gravitational/teleport/lib/reversetunnel"
"github.com/gravitational/teleport/lib/reversetunnelclient"
@@ -4321,8 +4321,7 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
var peerQUICTransport *quic.Transport
if !process.Config.Proxy.DisableReverseTunnel {
if listeners.proxyPeer != nil {
// TODO(espadolini): allow this when the implementation is merged
if false && os.Getenv("TELEPORT_UNSTABLE_QUIC_PROXY_PEERING") == "yes" {
if process.Config.Proxy.QUICProxyPeering {
// the stateless reset key is important in case there's a crash
// so peers can be told to close their side of the connections
// instead of having to wait for a timeout; for this reason, we
@@ -4707,7 +4706,7 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
var peerAddrString string
var peerServer *peer.Server
var peerQUICServer *peer.QUICServer
var peerQUICServer *peerquic.Server
if !process.Config.Proxy.DisableReverseTunnel && listeners.proxyPeer != nil {
peerAddr, err := process.Config.Proxy.PublicPeerAddr()
if err != nil {
@@ -4716,9 +4715,9 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
peerAddrString = peerAddr.String()
peerServer, err = peer.NewServer(peer.ServerConfig{
Log: process.logger,
ClusterDialer: clusterdial.NewClusterDialer(tsrv),
CipherSuites: cfg.CipherSuites,
Log: process.logger,
Dialer: reversetunnelclient.NewPeerDialer(tsrv),
CipherSuites: cfg.CipherSuites,
GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
return conn.serverGetCertificate()
},
@@ -4750,10 +4749,9 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
})
if peerQUICTransport != nil {
peerQUICServer, err := peer.NewQUICServer(peer.QUICServerConfig{
Log: process.logger,
ClusterDialer: clusterdial.NewClusterDialer(tsrv),
CipherSuites: cfg.CipherSuites,
peerQUICServer, err := peerquic.NewServer(peerquic.ServerConfig{
Log: process.logger,
Dialer: reversetunnelclient.NewPeerDialer(tsrv),
GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
return conn.serverGetCertificate()
},
@@ -4771,11 +4769,11 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
process.RegisterCriticalFunc("proxy.peer.quic", func() error {
if _, err := process.WaitForEvent(process.ExitContext(), ProxyReverseTunnelReady); err != nil {
logger.DebugContext(process.ExitContext(), "Process exiting: failed to start QUIC peer proxy service waiting for reverse tunnel server.")
logger.DebugContext(process.ExitContext(), "process exiting: failed to start QUIC peer proxy service waiting for reverse tunnel server")
return nil
}
logger.InfoContext(process.ExitContext(), "Starting QUIC peer proxy service.", "local_addr", logutils.StringerAttr(peerQUICTransport.Conn.LocalAddr()))
logger.InfoContext(process.ExitContext(), "starting QUIC peer proxy service", "local_addr", logutils.StringerAttr(peerQUICTransport.Conn.LocalAddr()))
err := peerQUICServer.Serve(peerQUICTransport)
if err != nil {
return trace.Wrap(err)
@@ -4797,8 +4795,8 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
logger.InfoContext(process.ExitContext(), "Enabling proxy group labels.", "group_id", cfg.Proxy.ProxyGroupID, "generation", cfg.Proxy.ProxyGroupGeneration)
}
if peerQUICTransport != nil {
staticLabels[types.ProxyPeerQUICLabel] = "x"
logger.InfoContext(process.ExitContext(), "Advertising proxy peering QUIC support.")
staticLabels[types.UnstableProxyPeerQUICLabel] = "yes"
logger.InfoContext(process.ExitContext(), "advertising proxy peering QUIC support")
}
sshProxy, err := regular.New(
lib/service/servicecfg/proxy.go
+5
View File
@@ -158,6 +158,11 @@ type ProxyConfig struct {
// proxy built-in version server to retrieve target versions. This is part
// of the automatic upgrades.
AutomaticUpgradesChannels automaticupgrades.Channels
// QUICProxyPeering will make it so that proxy peering will support inbound
// QUIC connections and will use QUIC to connect to peer proxies that
// advertise support for it.
QUICProxyPeering bool
}
// WebPublicAddr returns the address for the web endpoint on this proxy that