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

Author: jentfoo

lib/auth/grpcserver.go

@@ -70,6 +70,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"
@@ -5584,6 +5585,7 @@ func NewGRPCServer(cfg GRPCServerConfig) (*GRPCServer, error) {
 				PermitWithoutStream: true,
 			},
 		),
+		grpc.MaxConcurrentStreams(defaults.GRPCMaxConcurrentStreams),
 	)
 	if err != nil {
 		return nil, trace.Wrap(err)

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
 

lib/observability/tracing/collector.go

@@ -78,7 +78,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),
 	}

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)

lib/service/service.go

@@ -4283,6 +4283,7 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error {
 			otelgrpc.StreamServerInterceptor(),
 		),
 		grpc.Creds(creds),
+		grpc.MaxConcurrentStreams(defaults.GRPCMaxConcurrentStreams),
 	)
 
 	connMonitor, err := srv.NewConnectionMonitor(srv.ConnectionMonitorConfig{
@@ -5945,6 +5946,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)
@@ -6004,6 +6006,7 @@ func (process *TeleportProcess) initSecureGRPCServer(cfg initSecureGRPCServerCfg
 		grpc.ChainUnaryInterceptor(authMiddleware.UnaryInterceptors()...),
 		grpc.ChainStreamInterceptor(authMiddleware.StreamInterceptors()...),
 		grpc.Creds(creds),
+		grpc.MaxConcurrentStreams(defaults.GRPCMaxConcurrentStreams),
 	)
 
 	kubeServer, err := kubegrpc.New(kubegrpc.Config{

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.