From 0bc423ebd81bac56223b896626ddfbb2989ec12e Mon Sep 17 00:00:00 2001 From: Timothy Messier Date: Tue, 24 Oct 2023 14:52:25 +0000 Subject: [PATCH] feat(controller): Reload api rate limits on SIGHUP --- .../controller_ratelimit_reload_test.go | 327 ++++++++++++++++++ internal/cmd/commands/server/server.go | 13 + internal/daemon/controller/config.go | 6 + internal/daemon/controller/controller.go | 41 ++- internal/daemon/controller/handler.go | 2 +- internal/ratelimit/config.go | 6 + internal/ratelimit/handler.go | 10 +- internal/ratelimit/handler_test.go | 4 +- 8 files changed, 401 insertions(+), 8 deletions(-) create mode 100644 internal/cmd/commands/server/controller_ratelimit_reload_test.go diff --git a/internal/cmd/commands/server/controller_ratelimit_reload_test.go b/internal/cmd/commands/server/controller_ratelimit_reload_test.go new file mode 100644 index 000000000..61f4c7967 --- /dev/null +++ b/internal/cmd/commands/server/controller_ratelimit_reload_test.go @@ -0,0 +1,327 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package server + +import ( + "fmt" + "net/http" + "os" + "sync" + "testing" + "time" + + "github.com/hashicorp/boundary/internal/cmd/config" + "github.com/mitchellh/cli" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + ratelimitConfig = ` +disable_mlock = true + +telemetry { + prometheus_retention_time = "24h" + disable_hostname = true +} + +controller { + name = "test-controller" + description = "A default controller created for tests" + database { + url = "%s" + } + + api_rate_limit { + resources = ["*"] + actions = ["*"] + per = "total" + limit = 2 + period = "1m" + } +} + +kms "aead" { + purpose = "root" + aead_type = "aes-gcm" + key = "%s" + key_id = "global_root" +} + +kms "aead" { + purpose = "worker-auth" + aead_type = "aes-gcm" + key = "%s" + key_id = "global_worker-auth" +} + +kms "aead" { + purpose = "recovery" + aead_type = "aes-gcm" + key = "%s" + key_id = "global_recovery" +} + +listener "tcp" { + purpose = "api" + address = "127.0.0.1:9500" + tls_disable = true +} + +listener "tcp" { + address = "127.0.0.1:9501" + purpose = "cluster" +} +` + + ratelimitConfigReload = ` +disable_mlock = true + +telemetry { + prometheus_retention_time = "24h" + disable_hostname = true +} + +controller { + name = "test-controller" + description = "A default controller created for tests" + database { + url = "%s" + } + + api_rate_limit { + resources = ["*"] + actions = ["*"] + per = "total" + limit = 5 + period = "1m" + } +} + +kms "aead" { + purpose = "root" + aead_type = "aes-gcm" + key = "%s" + key_id = "global_root" +} + +kms "aead" { + purpose = "worker-auth" + aead_type = "aes-gcm" + key = "%s" + key_id = "global_worker-auth" +} + +kms "aead" { + purpose = "recovery" + aead_type = "aes-gcm" + key = "%s" + key_id = "global_recovery" +} + +listener "tcp" { + purpose = "api" + address = "127.0.0.1:9500" + tls_disable = true +} + +listener "tcp" { + address = "127.0.0.1:9501" + purpose = "cluster" +} +` +) + +func TestRealodControllerRateLimits(t *testing.T) { + td := t.TempDir() + + controllerKey := config.DevKeyGeneration() + + closeDB, url, _, err := getInitDatabase(t, controllerKey) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, closeDB()) }) + + cmd := testServerCommand(t, testServerCommandOpts{}) + + workerAuthKey := config.DevKeyGeneration() + recoveryKey := config.DevKeyGeneration() + cfgHcl := fmt.Sprintf(ratelimitConfig, url, controllerKey, workerAuthKey, recoveryKey) + require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644)) + + wg := &sync.WaitGroup{} + wg.Add(1) + go func() { + defer wg.Done() + + args := []string{"-config", td + "/config.hcl"} + exitCode := cmd.Run(args) + if exitCode != 0 { + output := cmd.UI.(*cli.MockUi).ErrorWriter.String() + cmd.UI.(*cli.MockUi).OutputWriter.String() + t.Errorf("got a non-zero exit status: %s", output) + } + }() + + // Wait until things are up and running (or timeout). + select { + case <-cmd.startedCh: + case <-time.After(15 * time.Second): + t.Fatal("timeout") + } + + // Change config so it is read for reloading + cfgHcl = fmt.Sprintf(ratelimitConfigReload, url, controllerKey, workerAuthKey, recoveryKey) + require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644)) + + c := http.Client{} + r, err := c.Do(func() *http.Request { + r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil) + require.NoError(t, err) + return r + }()) + require.NoError(t, err) + // unauthed request, so we expect a 400 + assert.Equal(t, http.StatusBadRequest, r.StatusCode) + assert.Equal(t, `limit=2, remaining=1, reset=59`, r.Header.Get("Ratelimit")) + assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy")) + + r, err = c.Do(func() *http.Request { + r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil) + require.NoError(t, err) + return r + }()) + require.NoError(t, err) + // unauthed request, so we expect a 400 + assert.Equal(t, http.StatusBadRequest, r.StatusCode) + assert.Equal(t, `limit=2, remaining=0, reset=59`, r.Header.Get("Ratelimit")) + assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy")) + + r, err = c.Do(func() *http.Request { + r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil) + require.NoError(t, err) + return r + }()) + require.NoError(t, err) + // out of quota, so we expect a 429 + assert.Equal(t, http.StatusTooManyRequests, r.StatusCode) + assert.Equal(t, `limit=2, remaining=0, reset=59`, r.Header.Get("Ratelimit")) + assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy")) + + cmd.SighupCh <- struct{}{} + select { + case <-cmd.reloadedCh: + case <-time.After(15 * time.Second): + t.Fatal("timeout") + } + + // Make another request, the limit should have reset and the new limit + // should get reported via the headers. + r, err = c.Do(func() *http.Request { + r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil) + require.NoError(t, err) + return r + }()) + require.NoError(t, err) + // unauthed request, so we expect a 400 + assert.Equal(t, http.StatusBadRequest, r.StatusCode) + assert.Equal(t, `limit=5, remaining=4, reset=59`, r.Header.Get("Ratelimit")) + assert.Equal(t, `5;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy")) + + cmd.ShutdownCh <- struct{}{} + wg.Wait() +} + +func TestRealodControllerRateLimitsSameConfig(t *testing.T) { + td := t.TempDir() + + // Create and migrate database A and B. + controllerKey := config.DevKeyGeneration() + + closeDB, url, _, err := getInitDatabase(t, controllerKey) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, closeDB()) }) + + cmd := testServerCommand(t, testServerCommandOpts{}) + + workerAuthKey := config.DevKeyGeneration() + recoveryKey := config.DevKeyGeneration() + cfgHcl := fmt.Sprintf(ratelimitConfig, url, controllerKey, workerAuthKey, recoveryKey) + require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644)) + + wg := &sync.WaitGroup{} + wg.Add(1) + go func() { + defer wg.Done() + + args := []string{"-config", td + "/config.hcl"} + exitCode := cmd.Run(args) + if exitCode != 0 { + output := cmd.UI.(*cli.MockUi).ErrorWriter.String() + cmd.UI.(*cli.MockUi).OutputWriter.String() + t.Errorf("got a non-zero exit status: %s", output) + } + }() + + // Wait until things are up and running (or timeout). + select { + case <-cmd.startedCh: + case <-time.After(15 * time.Second): + t.Fatal("timeout") + } + + c := http.Client{} + r, err := c.Do(func() *http.Request { + r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil) + require.NoError(t, err) + return r + }()) + require.NoError(t, err) + // unauthed request, so we expect a 400 + assert.Equal(t, http.StatusBadRequest, r.StatusCode) + assert.Equal(t, `limit=2, remaining=1, reset=59`, r.Header.Get("Ratelimit")) + assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy")) + + r, err = c.Do(func() *http.Request { + r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil) + require.NoError(t, err) + return r + }()) + require.NoError(t, err) + // unauthed request, so we expect a 400 + assert.Equal(t, http.StatusBadRequest, r.StatusCode) + assert.Equal(t, `limit=2, remaining=0, reset=59`, r.Header.Get("Ratelimit")) + assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy")) + + r, err = c.Do(func() *http.Request { + r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil) + require.NoError(t, err) + return r + }()) + require.NoError(t, err) + // out of quota, so we expect a 429 + assert.Equal(t, http.StatusTooManyRequests, r.StatusCode) + assert.Equal(t, `limit=2, remaining=0, reset=59`, r.Header.Get("Ratelimit")) + assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy")) + + cmd.SighupCh <- struct{}{} + select { + case <-cmd.reloadedCh: + case <-time.After(15 * time.Second): + t.Fatal("timeout") + } + + // Make another request, the limit should not have reset, and the current + // quota should still apply. + r, err = c.Do(func() *http.Request { + r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil) + require.NoError(t, err) + return r + }()) + require.NoError(t, err) + // should still be rate limited, so 429 + assert.Equal(t, http.StatusTooManyRequests, r.StatusCode) + assert.Equal(t, `limit=2, remaining=0, reset=59`, r.Header.Get("Ratelimit")) + assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy")) + + cmd.ShutdownCh <- struct{}{} + wg.Wait() +} diff --git a/internal/cmd/commands/server/server.go b/internal/cmd/commands/server/server.go index 19187e993..dec946830 100644 --- a/internal/cmd/commands/server/server.go +++ b/internal/cmd/commands/server/server.go @@ -622,6 +622,8 @@ func (c *Command) StartController(ctx context.Context) error { RawConfig: c.Config, Server: c.Server, TestOverrideWorkerAuthCaCertificateLifetime: c.flagWorkerAuthCaCertificateLifetime, + ApiRateLimits: c.Config.Controller.ApiRateLimits, + ApiRateLimiterMaxEntries: c.Config.Controller.ApiRateLimiterMaxEntries, } var err error @@ -841,6 +843,10 @@ func (c *Command) Reload(newConf *config.Config) error { reloadErrors = stderrors.Join(reloadErrors, fmt.Errorf("failed to reload controller database: %w", err)) } + if err := c.reloadControllerRateLimits(newConf); err != nil { + reloadErrors = stderrors.Join(reloadErrors, fmt.Errorf("failed to reload controller api rate limits: %w", err)) + } + if newConf != nil && c.worker != nil { workerReloadErr := func() error { if newConf.Controller != nil { @@ -952,6 +958,13 @@ func (c *Command) reloadControllerDatabase(newConfig *config.Config) error { return nil } +func (c *Command) reloadControllerRateLimits(newConfig *config.Config) error { + if c.controller == nil || newConfig == nil || newConfig.Controller == nil { + return nil + } + return c.controller.ReloadRateLimiter(newConfig.Controller.ApiRateLimits, newConfig.Controller.ApiRateLimiterMaxEntries) +} + // acquireSchemaManager returns a schema manager and generally acquires a shared lock on // the database. This is done as a mechanism to disallow running migration commands // while the database is in use. diff --git a/internal/daemon/controller/config.go b/internal/daemon/controller/config.go index 2e742f4a2..915ea5099 100644 --- a/internal/daemon/controller/config.go +++ b/internal/daemon/controller/config.go @@ -8,6 +8,7 @@ import ( "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/boundary/internal/cmd/config" + "github.com/hashicorp/boundary/internal/ratelimit" ) type Config struct { @@ -21,4 +22,9 @@ type Config struct { DisableAuthorizationFailures bool // Override worker auth CA certificate lifetime for testing TestOverrideWorkerAuthCaCertificateLifetime time.Duration + + // These are duplicated from the config.Config to track the state if these + // change via SIGHUP reloads. + ApiRateLimits ratelimit.Configs + ApiRateLimiterMaxEntries int } diff --git a/internal/daemon/controller/controller.go b/internal/daemon/controller/controller.go index ce25afa11..f6d6c2802 100644 --- a/internal/daemon/controller/controller.go +++ b/internal/daemon/controller/controller.go @@ -35,6 +35,7 @@ import ( kmsjob "github.com/hashicorp/boundary/internal/kms/job" "github.com/hashicorp/boundary/internal/plugin" "github.com/hashicorp/boundary/internal/plugin/loopback" + "github.com/hashicorp/boundary/internal/ratelimit" "github.com/hashicorp/boundary/internal/scheduler" "github.com/hashicorp/boundary/internal/scheduler/cleaner" "github.com/hashicorp/boundary/internal/scheduler/job" @@ -120,7 +121,7 @@ type Controller struct { apiGrpcServerListener grpcServerListener apiGrpcGatewayTicket string - rateLimiter *rate.Limiter + rateLimiter *atomic.Pointer[rate.Limiter] // Repo factory methods AuthTokenRepoFn common.AuthTokenRepoFactory @@ -172,6 +173,7 @@ func New(ctx context.Context, conf *Config) (*Controller, error) { pkiConnManager: cluster.NewDownstreamManager(), workerStatusGracePeriod: new(atomic.Int64), livenessTimeToStale: new(atomic.Int64), + rateLimiter: new(atomic.Pointer[rate.Limiter]), } if downstreamReceiverFactory != nil { @@ -253,10 +255,11 @@ func New(ctx context.Context, conf *Config) (*Controller, error) { return nil, fmt.Errorf("error parsing rate limit configuration: %w", err) } - c.rateLimiter, err = rate.NewLimiter(rateLimits, conf.RawConfig.Controller.ApiRateLimiterMaxEntries) + rateLimiter, err := rate.NewLimiter(rateLimits, conf.RawConfig.Controller.ApiRateLimiterMaxEntries) if err != nil { return nil, fmt.Errorf("error initializing rate limiter: %w", err) } + c.rateLimiter.Store(rateLimiter) var pluginLogger hclog.Logger for _, enabledPlugin := range c.enabledPlugins { @@ -631,3 +634,37 @@ func (c *Controller) Shutdown() error { func (c *Controller) WorkerStatusUpdateTimes() *sync.Map { return c.workerStatusUpdateTimes } + +func (c *Controller) getRateLimiter() *rate.Limiter { + return c.rateLimiter.Load() +} + +// ReloadRateLimiter replaces the Controller's rate.Limiter with a new rate.Limiter +// using the supplied ratelimit.Configs and max entries. If the configs and max +// entries match the current values, the rate.Limiter is not replaced and no +// error is returned. Otherwise the rate.Limiter is replaced, and the old rate.Limiter +// is shutdown. This means that a client's quota is effectively reset if the +// configuration changes. +func (c *Controller) ReloadRateLimiter(newLimitConfigs ratelimit.Configs, newMaxEntries int) error { + // Config has not changed, no need to reload. + if c.conf.ApiRateLimits.Equal(newLimitConfigs) && c.conf.ApiRateLimiterMaxEntries == newMaxEntries { + return nil + } + + ratelimits, err := newLimitConfigs.Limits(c.baseContext) + if err != nil { + return fmt.Errorf("error parsing rate limit configuration: %w", err) + } + + limiter, err := rate.NewLimiter(ratelimits, newMaxEntries) + if err != nil { + return err + } + + old := c.rateLimiter.Swap(limiter) + + c.conf.ApiRateLimits = newLimitConfigs + c.conf.ApiRateLimiterMaxEntries = newMaxEntries + + return old.Shutdown() +} diff --git a/internal/daemon/controller/handler.go b/internal/daemon/controller/handler.go index d6f1dbdb4..7ef1d4a08 100644 --- a/internal/daemon/controller/handler.go +++ b/internal/daemon/controller/handler.go @@ -74,7 +74,7 @@ func createMuxWithEndpoints(c *Controller, props HandlerProperties) (http.Handle } mux := http.NewServeMux() - mux.Handle("/v1/", ratelimit.Handler(c.baseContext, c.rateLimiter, grpcGwMux)) + mux.Handle("/v1/", ratelimit.Handler(c.baseContext, c.getRateLimiter, grpcGwMux)) mux.Handle(uiPath, handleUi(c)) isUiRequest := func(req *http.Request) bool { diff --git a/internal/ratelimit/config.go b/internal/ratelimit/config.go index a4ea23138..ade0571fe 100644 --- a/internal/ratelimit/config.go +++ b/internal/ratelimit/config.go @@ -6,6 +6,7 @@ package ratelimit import ( "context" "fmt" + "reflect" "sync" "time" @@ -108,6 +109,11 @@ type Config struct { // Configs is an ordered set of Config. type Configs []*Config +// Equal checks if a set of Configs is equal to another set of Configs. +func (c Configs) Equal(o Configs) bool { + return reflect.DeepEqual(c, o) +} + // Limits creates a slice of rate.Limit from the Configs. This will enumerate // every combination of resource+action, defining a Limit for each. func (c Configs) Limits(ctx context.Context) ([]*rate.Limit, error) { diff --git a/internal/ratelimit/handler.go b/internal/ratelimit/handler.go index 02c766812..5498fbaba 100644 --- a/internal/ratelimit/handler.go +++ b/internal/ratelimit/handler.go @@ -112,14 +112,18 @@ func extractResourceAction(path, method string) (res, act string, err error) { return } +// LimiterFunc returns a rate.Limiter +type LimiterFunc func() *rate.Limiter + // Handler is an http middleware handler that checks if a request is allowed -// using the provided rate limiter. If the request is allowed, the next handler +// using the rate limiter returned by f. If the request is allowed, the next handler // is called. Otherwise a 429 is returned with the Retry-After response header // set to the number of seconds the client should wait to make it's next request. -func Handler(ctx context.Context, l *rate.Limiter, next http.Handler) http.Handler { +func Handler(ctx context.Context, f LimiterFunc, next http.Handler) http.Handler { const op = "ratelimit.Handler" - return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + l := f() + reqInfo, ok := event.RequestInfoFromContext(req.Context()) if !ok || reqInfo == nil || reqInfo.ClientIp == "" { rw.WriteHeader(http.StatusInternalServerError) diff --git a/internal/ratelimit/handler_test.go b/internal/ratelimit/handler_test.go index 9bd0749bc..fd4883487 100644 --- a/internal/ratelimit/handler_test.go +++ b/internal/ratelimit/handler_test.go @@ -747,7 +747,7 @@ func TestHandler(t *testing.T) { next.ServeHTTP(rw, req) }) - }(ratelimit.Handler(ctx, tc.limiter, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + }(ratelimit.Handler(ctx, func() *rate.Limiter { return tc.limiter }, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) return }))), @@ -850,7 +850,7 @@ func TestHandlerErrors(t *testing.T) { next.ServeHTTP(rw, req) }) - }(ratelimit.Handler(ctx, r, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + }(ratelimit.Handler(ctx, func() *rate.Limiter { return r }, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) return }))),