Limit GRPC Active streams (#33936)

Originally there was a default limit of 100 max concurrent streams, however in 2017 the GRPC team removed this default: grpc/grpc-go#1624 With the recent HTTP/2 Rapid Reset DoS, it is now being encouraged to re-introduce a limit. The fix requires this value to be configured in fact: grpc/grpc-go#6703
gravitational · Oct 27, 2023 · 43a18f5 · 43a18f5
1 parent 151cef3
commit 43a18f5
Show file tree

Hide file tree

Showing 6 changed files with 18 additions and 2 deletions.
diff --git a/lib/auth/grpcserver.go b/lib/auth/grpcserver.go
@@ -58,6 +58,7 @@ import (
 	wanlib "github.com/gravitational/teleport/lib/auth/webauthn"
 	"github.com/gravitational/teleport/lib/authz"
 	"github.com/gravitational/teleport/lib/backend"
+	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/events"
 	"github.com/gravitational/teleport/lib/httplib"
 	"github.com/gravitational/teleport/lib/joinserver"
@@ -5128,6 +5129,7 @@ func NewGRPCServer(cfg GRPCServerConfig) (*GRPCServer, error) {
 				PermitWithoutStream: true,
 			},
 		),
+		grpc.MaxConcurrentStreams(defaults.GRPCMaxConcurrentStreams),
 	}
 	server := grpc.NewServer(opts...)
 	authServer := &GRPCServer{

diff --git a/lib/defaults/defaults.go b/lib/defaults/defaults.go
@@ -100,6 +100,10 @@ const (
 	// By default all users use /bin/bash
 	DefaultShell = "/bin/bash"
 
+	// GRPCMaxConcurrentStreams is the max GRPC streams that can be active at a time.  Once the limit is reached new
+	// RPC calls will queue until capacity is available.
+	GRPCMaxConcurrentStreams = 1000
+
 	// HTTPMaxIdleConns is the max idle connections across all hosts.
 	HTTPMaxIdleConns = 2000
 

diff --git a/lib/observability/tracing/collector.go b/lib/observability/tracing/collector.go
@@ -31,6 +31,8 @@ import (
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/protobuf/proto"
+
+	"github.com/gravitational/teleport/lib/defaults"
 )
 
 // Collector is a simple in memory implementation of an OpenTelemetry Collector
@@ -75,7 +77,7 @@ func NewCollector(cfg CollectorConfig) (*Collector, error) {
 	c := &Collector{
 		grpcLn:     grpcLn,
 		httpLn:     httpLn,
-		grpcServer: grpc.NewServer(grpc.Creds(creds)),
+		grpcServer: grpc.NewServer(grpc.Creds(creds), grpc.MaxConcurrentStreams(defaults.GRPCMaxConcurrentStreams)),
 		tlsConfing: tlsConfig,
 		exportedC:  make(chan struct{}, 1),
 	}

diff --git a/lib/proxy/peer/server.go b/lib/proxy/peer/server.go
@@ -31,6 +31,7 @@ import (
 	"github.com/gravitational/teleport/api/metadata"
 	"github.com/gravitational/teleport/api/utils/grpc/interceptors"
 	"github.com/gravitational/teleport/lib/auth"
+	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/utils"
 )
 
@@ -141,6 +142,7 @@ func NewServer(config ServerConfig) (*Server, error) {
 			MinTime:             peerKeepAlive,
 			PermitWithoutStream: true,
 		}),
+		grpc.MaxConcurrentStreams(defaults.GRPCMaxConcurrentStreams),
 	)
 
 	proto.RegisterProxyServiceServer(server, config.service)

diff --git a/lib/service/service.go b/lib/service/service.go
@@ -3878,6 +3878,7 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
 			otelgrpc.StreamServerInterceptor(),
 		),
 		grpc.Creds(credentials.NewTLS(serverTLSConfig)),
+		grpc.MaxConcurrentStreams(defaults.GRPCMaxConcurrentStreams),
 	)
 
 	process.RegisterCriticalFunc("proxy.ssh", func() error {
@@ -5394,6 +5395,7 @@ func (process *TeleportProcess) initPublicGRPCServer(
 			// available for some time.
 			MaxConnectionIdle: 10 * time.Second,
 		}),
+		grpc.MaxConcurrentStreams(defaults.GRPCMaxConcurrentStreams),
 	)
 	joinServiceServer := joinserver.NewJoinServiceGRPCServer(conn.Client)
 	proto.RegisterJoinServiceServer(server, joinServiceServer)
@@ -5443,6 +5445,7 @@ func (process *TeleportProcess) initSecureGRPCServer(cfg initSecureGRPCServerCfg
 		grpc.Creds(credentials.NewTLS(
 			copyAndConfigureTLS(serverTLSConfig, process.log, cfg.accessPoint, clusterName),
 		)),
+		grpc.MaxConcurrentStreams(defaults.GRPCMaxConcurrentStreams),
 	)
 
 	kubeServer, err := kubegrpc.New(kubegrpc.Config{

diff --git a/lib/teleterm/apiserver/apiserver.go b/lib/teleterm/apiserver/apiserver.go
@@ -23,6 +23,7 @@ import (
 	"google.golang.org/grpc"
 
 	api "github.com/gravitational/teleport/gen/proto/go/teleport/lib/teleterm/v1"
+	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/teleterm/apiserver/handler"
 	"github.com/gravitational/teleport/lib/utils"
 )
@@ -41,7 +42,9 @@ func New(cfg Config) (*APIServer, error) {
 	}
 
 	grpcServer := grpc.NewServer(cfg.TshdServerCreds,
-		grpc.ChainUnaryInterceptor(withErrorHandling(cfg.Log)))
+		grpc.ChainUnaryInterceptor(withErrorHandling(cfg.Log)),
+		grpc.MaxConcurrentStreams(defaults.GRPCMaxConcurrentStreams),
+	)
 
 	// Create Terminal service.