lib/limiter: remove the Oxy rate limiter (#47257)

I've vendored in parts of gravitational/oxy under the
lib/limiter/internal/ratelimit package, taking the opportunity to
convert to slog and replace mailgun dependencies (ttlmap, timetools)
with Teleport equivalents.

This commit also removes the max_users limiter option.
This setting never actually did what it claims to do. The max_users
option was passed in as a capacity to a TTL map, but the TTL map
doesn't reject new entries when the capacity is exceeded, it
just removes old entries to make room.

Removing old entries from the TTL map did not enforce any sort of
user limit - it has the opposite effect. It "resets" existing limits
for the old user, effectively un-limiting them!

In practice, we've never actually been able to enforce a maximum
number of simultaneous users, so rather than changing the behavior
I've opted to remove the field from everything other than the config
file.

Makefile

@@ -1303,6 +1303,7 @@ ADDLICENSE_COMMON_ARGS := -c 'Gravitational, Inc.' \
 		-ignore 'version.go' \
 		-ignore 'web/packages/design/src/assets/icomoon/style.css' \
 		-ignore 'web/packages/teleport/src/ironrdp/**' \
+		-ignore 'lib/limiter/internal/ratelimit/**' \
 		-ignore 'webassets/**' \
 		-ignore 'ignoreme'
 ADDLICENSE_AGPL3_ARGS := $(ADDLICENSE_COMMON_ARGS) \

docs/pages/admin-guides/deploy-a-cluster/deployments/gcp.mdx

@@ -88,7 +88,7 @@ teleport:
 
 <Admonition type="warning" title="Table Names">
 Be careful to ensure that `Example_FIRESTORE_CLUSTER_STATE` and `Example_FIRESTORE_AUDIT_LOGS`
-refer to *different* Firestore collections. The schema is different for each, and using the same 
+refer to *different* Firestore collections. The schema is different for each, and using the same
 collection for both types of data will result in errors.
 </Admonition>
 
@@ -232,9 +232,6 @@ teleport:
   nodename: teleport-auth-server
   data_dir: /var/lib/teleport
   pid_file: /run/teleport.pid
-  connection_limits:
-    max_connections: 15000
-    max_users: 250
   log:
     output: stderr
     severity: DEBUG
@@ -270,9 +267,6 @@ teleport:
   nodename: teleport-auth-server
   data_dir: /var/lib/teleport
   pid_file: /run/teleport.pid
-  connection_limits:
-    max_connections: 15000
-    max_users: 250
   log:
     output: stderr
     severity: DEBUG
@@ -338,8 +332,8 @@ Save the following configuration file as `/etc/teleport.yaml` on the Node:
 version: v3
 teleport:
   auth_token: (= presets.tokens.second =)
-  # Teleport agents can be joined to the cluster via the Proxy Service's 
-  # public address. This will establish a reverse tunnel between the Proxy 
+  # Teleport agents can be joined to the cluster via the Proxy Service's
+  # public address. This will establish a reverse tunnel between the Proxy
   # Service and the agent that is used for all traffic.
   proxy_server: teleport.example.com:443
 # enable the SSH Service and disable the Auth and Proxy Services

docs/pages/admin-guides/management/operations/scaling.mdx

@@ -32,7 +32,6 @@ to `65000`.
 teleport:
   connection_limits:
     max_connections: 65000
-    max_users: 1000
 ```
 
 ## Agent configuration

docs/pages/includes/config-reference/instance-wide.yaml

@@ -96,11 +96,11 @@ teleport:
     #  # The cache is enabled by default, it can be disabled with this flag
     #  enabled: true
 
-    # Teleport throttles all connections to avoid abuse. These settings allow
-    # you to adjust the default limits
+    # Teleport can limit the number of connections coming from each client
+    # IP address to avoid abuse. Note that these limits are enforced separately
+    # for each service (SSH, Kubernetes, etc.)
     connection_limits:
         max_connections: 1000
-        max_users: 250
 
     # Logging configuration. Possible output values to disk via
     # '/var/lib/teleport/teleport.log',

go.mod

@@ -143,8 +143,6 @@ require (
 	github.com/keys-pub/go-libfido2 v1.5.3-0.20220306005615-8ab03fb1ec27 // replaced
 	github.com/lib/pq v1.10.9
 	github.com/mailgun/mailgun-go/v4 v4.16.0
-	github.com/mailgun/timetools v0.0.0-20170619190023-f3a7b8ffff47
-	github.com/mailgun/ttlmap v0.0.0-20170619185759-c1c17f74874f
 	github.com/mattn/go-sqlite3 v1.14.23
 	github.com/mdlayher/netlink v1.7.2
 	github.com/microsoft/go-mssqldb v1.7.2 // replaced
@@ -424,6 +422,8 @@ require (
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mailgun/errors v0.3.0 // indirect
+	github.com/mailgun/timetools v0.0.0-20170619190023-f3a7b8ffff47 // indirect
+	github.com/mailgun/ttlmap v0.0.0-20170619185759-c1c17f74874f // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattermost/xml-roundtrip-validator v0.1.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect

lib/auth/helpers.go

@@ -883,8 +883,7 @@ func (cfg *TestTLSServerConfig) CheckAndSetDefaults() error {
 	// use very permissive limiter configuration by default
 	if cfg.Limiter == nil {
 		cfg.Limiter = &limiter.Config{
-			MaxConnections:   1000,
-			MaxNumberOfUsers: 1000,
+			MaxConnections: 1000,
 		}
 	}
 	return nil

lib/auth/middleware.go

@@ -33,7 +33,6 @@ import (
 	"time"
 
 	"github.com/coreos/go-semver/semver"
-	"github.com/gravitational/oxy/ratelimit"
 	"github.com/gravitational/trace"
 	grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
 	"github.com/prometheus/client_golang/prometheus"
@@ -361,7 +360,7 @@ func (a *Middleware) Wrap(h http.Handler) {
 	a.Handler = h
 }
 
-func getCustomRate(endpoint string) *ratelimit.RateSet {
+func getCustomRate(endpoint string) *limiter.RateSet {
 	switch endpoint {
 	// Account recovery RPCs.
 	case
@@ -370,7 +369,7 @@ func getCustomRate(endpoint string) *ratelimit.RateSet {
 		"/proto.AuthService/GetAccountRecoveryToken",
 		"/proto.AuthService/StartAccountRecovery",
 		"/proto.AuthService/VerifyAccountRecovery":
-		rates := ratelimit.NewRateSet()
+		rates := limiter.NewRateSet()
 		// This limit means: 1 request per minute with bursts up to 10 requests.
 		if err := rates.Add(time.Minute, 1, 10); err != nil {
 			log.WithError(err).Debugf("Failed to define a custom rate for rpc method %q, using default rate", endpoint)
@@ -382,7 +381,7 @@ func getCustomRate(endpoint string) *ratelimit.RateSet {
 		const period = defaults.LimiterPeriod
 		const average = defaults.LimiterAverage
 		const burst = defaults.LimiterBurst
-		rates := ratelimit.NewRateSet()
+		rates := limiter.NewRateSet()
 		if err := rates.Add(period, average, burst); err != nil {
 			log.WithError(err).Debugf("Failed to define a custom rate for rpc method %q, using default rate", endpoint)
 			return nil

lib/config/configuration.go

@@ -633,9 +633,7 @@ func ApplyFileConfig(fc *FileConfig, cfg *servicecfg.Config) error {
 		if fc.Limits.MaxConnections > 0 {
 			l.MaxConnections = fc.Limits.MaxConnections
 		}
-		if fc.Limits.MaxUsers > 0 {
-			l.MaxNumberOfUsers = fc.Limits.MaxUsers
-		}
+
 		for _, rate := range fc.Limits.Rates {
 			l.Rates = append(l.Rates, limiter.Rate{
 				Period:  rate.Period,

lib/config/configuration_test.go

@@ -1827,8 +1827,7 @@ func TestSetDefaultListenerAddresses(t *testing.T) {
 					Enabled: false,
 				},
 				Limiter: limiter.Config{
-					MaxConnections:   defaults.LimiterMaxConnections,
-					MaxNumberOfUsers: 250,
+					MaxConnections: defaults.LimiterMaxConnections,
 				},
 				IdP: servicecfg.IdP{
 					SAMLIdP: servicecfg.SAMLIdP{
@@ -1855,8 +1854,7 @@ func TestSetDefaultListenerAddresses(t *testing.T) {
 					Enabled: true,
 				},
 				Limiter: limiter.Config{
-					MaxConnections:   defaults.LimiterMaxConnections,
-					MaxNumberOfUsers: 250,
+					MaxConnections: defaults.LimiterMaxConnections,
 				},
 				IdP: servicecfg.IdP{
 					SAMLIdP: servicecfg.SAMLIdP{
@@ -2177,8 +2175,7 @@ func TestProxyConfigurationVersion(t *testing.T) {
 					Enabled: true,
 				},
 				Limiter: limiter.Config{
-					MaxConnections:   defaults.LimiterMaxConnections,
-					MaxNumberOfUsers: 250,
+					MaxConnections: defaults.LimiterMaxConnections,
 				},
 				IdP: servicecfg.IdP{
 					SAMLIdP: servicecfg.SAMLIdP{
@@ -2206,8 +2203,7 @@ func TestProxyConfigurationVersion(t *testing.T) {
 					Enabled: true,
 				},
 				Limiter: limiter.Config{
-					MaxConnections:   defaults.LimiterMaxConnections,
-					MaxNumberOfUsers: 250,
+					MaxConnections: defaults.LimiterMaxConnections,
 				},
 				IdP: servicecfg.IdP{
 					SAMLIdP: servicecfg.SAMLIdP{
@@ -4119,8 +4115,7 @@ func TestApplyKubeConfig(t *testing.T) {
 					},
 				},
 				Limiter: limiter.Config{
-					MaxConnections:   defaults.LimiterMaxConnections,
-					MaxNumberOfUsers: 250,
+					MaxConnections: defaults.LimiterMaxConnections,
 				},
 			},
 		},
@@ -4167,8 +4162,7 @@ func TestApplyKubeConfig(t *testing.T) {
 					},
 				},
 				Limiter: limiter.Config{
-					MaxConnections:   defaults.LimiterMaxConnections,
-					MaxNumberOfUsers: 250,
+					MaxConnections: defaults.LimiterMaxConnections,
 				},
 			},
 		},

lib/config/fileconf.go

@@ -512,8 +512,10 @@ type ConnectionRate struct {
 // ConnectionLimits sets up connection limiter
 type ConnectionLimits struct {
 	MaxConnections int64            `yaml:"max_connections"`
-	MaxUsers       int              `yaml:"max_users"`
 	Rates          []ConnectionRate `yaml:"rates,omitempty"`
+
+	// Deprecated: MaxUsers has no effect.
+	MaxUsers int `yaml:"max_users"`
 }
 
 // LegacyLog contains the old format of the 'format' field

lib/defaults/defaults.go

@@ -371,9 +371,6 @@ const (
 	// LimiterMaxConnections Number of max. simultaneous connections to a service
 	LimiterMaxConnections = 15000
 
-	// LimiterMaxConcurrentUsers Number of max. simultaneous connected users/logins
-	LimiterMaxConcurrentUsers = 250
-
 	// LimiterMaxConcurrentSignatures limits maximum number of concurrently
 	// generated signatures by the auth server
 	LimiterMaxConcurrentSignatures = 10
@@ -626,7 +623,6 @@ const (
 // ConfigureLimiter assigns the default parameters to a connection throttler (AKA limiter)
 func ConfigureLimiter(lc *limiter.Config) {
 	lc.MaxConnections = LimiterMaxConnections
-	lc.MaxNumberOfUsers = LimiterMaxConcurrentUsers
 }
 
 // AuthListenAddr returns the default listening address for the Auth service

lib/kube/proxy/utils_testing.go

@@ -142,8 +142,7 @@ func SetupTestContext(ctx context.Context, t *testing.T, cfg TestConfig) *TestCo
 		// this issue.
 		auth.WithLimiterConfig(
 			&limiter.Config{
-				MaxConnections:   100000,
-				MaxNumberOfUsers: 1000,
+				MaxConnections: 100000,
 			},
 		),
 	)
@@ -273,8 +272,7 @@ func SetupTestContext(ctx context.Context, t *testing.T, cfg TestConfig) *TestCo
 		TLS:           kubeServiceTLSConfig.Clone(),
 		AccessPoint:   client,
 		LimiterConfig: limiter.Config{
-			MaxConnections:   1000,
-			MaxNumberOfUsers: 1000,
+			MaxConnections: 1000,
 		},
 		// each time heartbeat is called we insert data into the channel.
 		// this is used to make sure that heartbeat started and the clusters
@@ -357,8 +355,7 @@ func SetupTestContext(ctx context.Context, t *testing.T, cfg TestConfig) *TestCo
 		AccessPoint:              client,
 		KubernetesServersWatcher: kubeServersWatcher,
 		LimiterConfig: limiter.Config{
-			MaxConnections:   1000,
-			MaxNumberOfUsers: 1000,
+			MaxConnections: 1000,
 		},
 		Log:             log,
 		InventoryHandle: inventoryHandle,

lib/limiter/connlimiter.go

@@ -22,12 +22,12 @@ import (
 	"context"
 	"log/slog"
 	"net/http"
-	"strings"
 	"sync"
 
 	"github.com/gravitational/trace"
 
 	"github.com/gravitational/teleport"
+	"github.com/gravitational/teleport/lib/limiter/internal/ratelimit"
 )
 
 // ConnectionsLimiter is a network connection limiter.
@@ -63,23 +63,21 @@ func (l *ConnectionsLimiter) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// TODO: use net.SplitHostPort to be more compatible with IPv6
-	token, _, _ := strings.Cut(r.RemoteAddr, ":")
-	if token == "" {
+	clientIP, err := ratelimit.ExtractClientIP(r)
+	if err != nil {
 		l.log.WarnContext(context.Background(), "failed to extract source IP", "remote_addr", r.RemoteAddr)
-		sc := http.StatusInternalServerError
-		http.Error(w, http.StatusText(sc), sc)
+		ratelimit.ServeHTTPError(w, r, err)
 		return
 	}
 
-	if err := l.AcquireConnection(token); err != nil {
-		l.log.InfoContext(context.Background(), "limiting request", "token", token, "error", err)
+	if err := l.AcquireConnection(clientIP); err != nil {
+		l.log.InfoContext(context.Background(), "limiting request", "token", clientIP, "error", err)
 		w.WriteHeader(http.StatusTooManyRequests)
 		w.Write([]byte(trace.UserMessage(err)))
 		return
 	}
 
-	defer l.ReleaseConnection(token)
+	defer l.ReleaseConnection(clientIP)
 
 	l.next.ServeHTTP(w, r)
 }

lib/limiter/internal/ratelimit/README.md

@@ -0,0 +1,4 @@
+# ratelimit
+
+The code in this directory is derived from https://github.com/gravitational/oxy
+and is available under an Apache 2.0 license.

lib/limiter/internal/ratelimit/http.go

@@ -0,0 +1,60 @@
+/*
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package ratelimit
+
+import (
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"strconv"
+	"strings"
+)
+
+func ServeHTTPError(w http.ResponseWriter, r *http.Request, err error) {
+	var le *limitExceededError
+	if errors.As(err, &le) {
+		// https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After#delay-seconds
+		w.Header().Set("Retry-After", strconv.Itoa(int(le.delay.Seconds())+1))
+		w.WriteHeader(http.StatusTooManyRequests)
+		w.Write([]byte(err.Error()))
+		return
+	}
+
+	status := http.StatusInternalServerError
+	var netError net.Error
+
+	if errors.Is(err, io.EOF) {
+		status = http.StatusBadGateway
+	} else if errors.As(err, &netError) {
+		if netError.Timeout() {
+			status = http.StatusGatewayTimeout
+		} else {
+			status = http.StatusBadGateway
+		}
+	}
+
+	http.Error(w, http.StatusText(status), status)
+}
+
+func ExtractClientIP(r *http.Request) (string, error) {
+	// TODO: use net.SplitHostPort to be compatible with IPv6
+	token, _, _ := strings.Cut(r.RemoteAddr, ":")
+	if token == "" {
+		return "", errors.New("failed to extract source IP")
+	}
+
+	return token, nil
+}