diff --git a/config/config.go b/config/config.go index 4570e051..4397499d 100644 --- a/config/config.go +++ b/config/config.go @@ -212,13 +212,33 @@ type EventsConfig struct { // variables, individual fields are not documented here; instead, see the `README.md` section on // configuration. type RedisConfig struct { - Host string `conf:"REDIS_HOST"` - Port ct.OptIntGreaterThanZero - URL ct.OptURLAbsolute `conf:"REDIS_URL"` - LocalTTL ct.OptDuration `conf:"CACHE_TTL"` - TLS bool `conf:"REDIS_TLS"` - Username string `conf:"REDIS_USERNAME"` - Password string `conf:"REDIS_PASSWORD"` + Host string `conf:"REDIS_HOST"` + Port ct.OptIntGreaterThanZero + URL ct.OptURLAbsolute `conf:"REDIS_URL"` + LocalTTL ct.OptDuration `conf:"CACHE_TTL"` + TLS bool `conf:"REDIS_TLS"` + Username string `conf:"REDIS_USERNAME"` + Password string `conf:"REDIS_PASSWORD"` + AWSAuth bool `conf:"REDIS_AWS_AUTH"` + AWSCacheName string `conf:"REDIS_AWS_CACHE_NAME"` + // AWSRegion overrides the AWS region used for SigV4 signing of ElastiCache IAM + // authentication tokens. If empty, the region is resolved from the standard AWS + // credential chain (AWS_DEFAULT_REGION env var, ~/.aws/config, IMDS, etc.). + // Only meaningful when AWSAuth is true; ignored otherwise. + AWSRegion string `conf:"REDIS_AWS_REGION"` + // AWSServerless indicates that the target ElastiCache cache is a Serverless cache. + // When true, the SigV4 presigned token includes the query parameter + // ResourceType=ServerlessCache, which is required by ElastiCache Serverless for IAM + // authentication. Without this flag, authentication against a Serverless cache fails + // with an opaque WRONGPASS error. Only meaningful when AWSAuth is true; ignored otherwise. + // + // EXPERIMENTAL / NOT FUNCTIONALLY SUPPORTED: this flag covers only the IAM auth + // handshake. ElastiCache Serverless is cluster-mode-only, and Relay's Redis clients are + // not cluster-aware, so multi-key operations (SDK data-store writes, big-segments + // Watch/TxPipelined transactions) will fail with CROSSSLOT. Full Serverless support is + // blocked on cluster-aware client work that has not shipped. Do not rely on this in + // production. See docs/persistent-storage.md. + AWSServerless bool `conf:"REDIS_AWS_SERVERLESS"` } // ConsulConfig configures the optional Consul integration. diff --git a/config/config_validation.go b/config/config_validation.go index 4c528c87..aef91c4a 100644 --- a/config/config_validation.go +++ b/config/config_validation.go @@ -21,6 +21,10 @@ var ( ` if using DynamoDB, table name) must be specified and must contain "` + AutoConfigEnvironmentIDPlaceholder + `"`) errRedisURLWithHostAndPort = errors.New("please specify Redis URL or host/port, but not both") errRedisBadHostname = errors.New("invalid Redis hostname") + errRedisAWSAuthRequiresTLS = errors.New("REDIS_AWS_AUTH requires REDIS_TLS=true (ElastiCache IAM authentication only works over TLS)") + errRedisAWSAuthRequiresUsername = errors.New("REDIS_AWS_AUTH requires REDIS_USERNAME") + errRedisAWSAuthRequiresCacheName = errors.New("REDIS_AWS_AUTH requires REDIS_AWS_CACHE_NAME") + errRedisAWSAuthForbidsPassword = errors.New("REDIS_AWS_AUTH cannot be combined with REDIS_PASSWORD (mutually exclusive)") errConsulTokenAndTokenFile = errors.New("Consul token must be specified as either an inline value or a file, but not both") //nolint:staticcheck errAutoConfWithFilters = errors.New("cannot configure filters if auto-configuration is enabled") errCacheKeyWithoutStore = errors.New("AUTO_CONFIG_CACHE_KEY requires Redis or DynamoDB to be enabled") @@ -308,4 +312,30 @@ func normalizeRedisConfig(result *ct.ValidationResult, c *Config) { c.Redis.Host = "" c.Redis.Port = ct.OptIntGreaterThanZero{} } + + if c.Redis.AWSAuth { + if !c.Redis.TLS { + result.AddError(nil, errRedisAWSAuthRequiresTLS) + } + if c.Redis.Username == "" { + result.AddError(nil, errRedisAWSAuthRequiresUsername) + } + if c.Redis.AWSCacheName == "" { + result.AddError(nil, errRedisAWSAuthRequiresCacheName) + } + if c.Redis.Password != "" { + result.AddError(nil, errRedisAWSAuthForbidsPassword) + } + // A password embedded in the URL (e.g. rediss://user:secret@host) would + // also bypass IAM auth and get sent to ElastiCache as a static AUTH — + // reject it for the same reason as REDIS_PASSWORD. + if u := c.Redis.URL.Get(); u != nil && u.User != nil { + if _, hasPassword := u.User.Password(); hasPassword { + result.AddError(nil, errRedisAWSAuthForbidsPassword) + } + } + if c.Redis.AWSCacheName != "" { + c.Redis.AWSCacheName = strings.ToLower(c.Redis.AWSCacheName) + } + } } diff --git a/config/test_data_configs_invalid_test.go b/config/test_data_configs_invalid_test.go index 9d583bbc..da04b2cb 100644 --- a/config/test_data_configs_invalid_test.go +++ b/config/test_data_configs_invalid_test.go @@ -37,6 +37,11 @@ func makeInvalidConfigs() []testDataInvalidConfig { makeInvalidConfigRedisConflictingParams(), makeInvalidConfigRedisNoPrefix(), makeInvalidConfigRedisAutoConfNoPrefix(), + makeInvalidConfigRedisAWSAuthWithoutTLS(), + makeInvalidConfigRedisAWSAuthWithoutUsername(), + makeInvalidConfigRedisAWSAuthWithoutCacheName(), + makeInvalidConfigRedisAWSAuthWithPassword(), + makeInvalidConfigRedisAWSAuthWithURLEmbeddedPassword(), makeInvalidConfigConsulNoPrefix(), makeInvalidConfigConsulAutoConfNoPrefix(), makeInvalidConfigConsulTokenAndTokenFile(), @@ -467,3 +472,109 @@ Enabled = true ` return c } + +func makeInvalidConfigRedisAWSAuthWithoutTLS() testDataInvalidConfig { + c := testDataInvalidConfig{name: "Redis - AWS auth without TLS"} + c.envVarsError = errRedisAWSAuthRequiresTLS.Error() + c.envVars = map[string]string{ + "USE_REDIS": "1", + "REDIS_URL": "redis://my-cluster.amazonaws.com:6379", + "REDIS_AWS_AUTH": "1", + "REDIS_USERNAME": "iam-user", + "REDIS_AWS_CACHE_NAME": "my-cache", + } + c.fileContent = ` +[Redis] +Url = "redis://my-cluster.amazonaws.com:6379" +AWSAuth = true +Username = "iam-user" +AWSCacheName = "my-cache" +` + return c +} + +func makeInvalidConfigRedisAWSAuthWithoutUsername() testDataInvalidConfig { + c := testDataInvalidConfig{name: "Redis - AWS auth without username"} + c.envVarsError = errRedisAWSAuthRequiresUsername.Error() + c.envVars = map[string]string{ + "USE_REDIS": "1", + "REDIS_URL": "rediss://my-cluster.amazonaws.com:6379", + "REDIS_TLS": "1", + "REDIS_AWS_AUTH": "1", + "REDIS_AWS_CACHE_NAME": "my-cache", + } + c.fileContent = ` +[Redis] +Url = "rediss://my-cluster.amazonaws.com:6379" +TLS = true +AWSAuth = true +AWSCacheName = "my-cache" +` + return c +} + +func makeInvalidConfigRedisAWSAuthWithoutCacheName() testDataInvalidConfig { + c := testDataInvalidConfig{name: "Redis - AWS auth without cache name"} + c.envVarsError = errRedisAWSAuthRequiresCacheName.Error() + c.envVars = map[string]string{ + "USE_REDIS": "1", + "REDIS_URL": "rediss://my-cluster.amazonaws.com:6379", + "REDIS_TLS": "1", + "REDIS_AWS_AUTH": "1", + "REDIS_USERNAME": "iam-user", + } + c.fileContent = ` +[Redis] +Url = "rediss://my-cluster.amazonaws.com:6379" +TLS = true +AWSAuth = true +Username = "iam-user" +` + return c +} + +func makeInvalidConfigRedisAWSAuthWithPassword() testDataInvalidConfig { + c := testDataInvalidConfig{name: "Redis - AWS auth with password"} + c.envVarsError = errRedisAWSAuthForbidsPassword.Error() + c.envVars = map[string]string{ + "USE_REDIS": "1", + "REDIS_URL": "rediss://my-cluster.amazonaws.com:6379", + "REDIS_TLS": "1", + "REDIS_AWS_AUTH": "1", + "REDIS_USERNAME": "iam-user", + "REDIS_AWS_CACHE_NAME": "my-cache", + "REDIS_PASSWORD": "should-not-be-here", + } + c.fileContent = ` +[Redis] +Url = "rediss://my-cluster.amazonaws.com:6379" +TLS = true +AWSAuth = true +Username = "iam-user" +AWSCacheName = "my-cache" +Password = "should-not-be-here" +` + return c +} + +func makeInvalidConfigRedisAWSAuthWithURLEmbeddedPassword() testDataInvalidConfig { + c := testDataInvalidConfig{name: "Redis - AWS auth with password embedded in URL"} + c.envVarsError = errRedisAWSAuthForbidsPassword.Error() + c.envVars = map[string]string{ + "USE_REDIS": "1", + "REDIS_URL": "rediss://iam-user:embedded-pw@my-cluster.amazonaws.com:6379", + "REDIS_TLS": "1", + "REDIS_AWS_AUTH": "1", + "REDIS_USERNAME": "iam-user", + "REDIS_AWS_CACHE_NAME": "my-cache", + } + c.fileContent = ` +[Redis] +Url = "rediss://iam-user:embedded-pw@my-cluster.amazonaws.com:6379" +TLS = true +AWSAuth = true +Username = "iam-user" +AWSCacheName = "my-cache" +` + return c +} diff --git a/config/test_data_configs_valid_test.go b/config/test_data_configs_valid_test.go index 72eb0fa0..9c70e48b 100644 --- a/config/test_data_configs_valid_test.go +++ b/config/test_data_configs_valid_test.go @@ -78,6 +78,10 @@ func makeValidConfigs() []testDataValidConfig { makeValidConfigRedisPortOnly(), makeValidConfigRedisDockerPort(), makeValidConfigRedisOneEnvNoPrefix(), + makeValidConfigRedisAWSAuth(), + makeValidConfigRedisAWSAuthLowercasesCacheName(), + makeValidConfigRedisAWSAuthWithRegion(), + makeValidConfigRedisAWSAuthServerless(), makeValidConfigConsulMinimal(), makeValidConfigConsulAll(), makeValidConfigConsulOneEnvNoPrefix(), @@ -534,6 +538,132 @@ Host = localhost return c } +func makeValidConfigRedisAWSAuth() testDataValidConfig { + c := testDataValidConfig{name: "Redis - AWS IAM auth happy path"} + c.makeConfig = func(c *Config) { + c.Redis = RedisConfig{ + URL: newOptURLAbsoluteMustBeValid("rediss://my-cluster.amazonaws.com:6379"), + TLS: true, + Username: "iam-user", + AWSAuth: true, + AWSCacheName: "my-cache", + } + } + c.envVars = map[string]string{ + "USE_REDIS": "1", + "REDIS_URL": "rediss://my-cluster.amazonaws.com:6379", + "REDIS_TLS": "1", + "REDIS_USERNAME": "iam-user", + "REDIS_AWS_AUTH": "1", + "REDIS_AWS_CACHE_NAME": "my-cache", + } + c.fileContent = ` +[Redis] +Url = "rediss://my-cluster.amazonaws.com:6379" +TLS = true +Username = "iam-user" +AWSAuth = true +AWSCacheName = "my-cache" +` + return c +} + +func makeValidConfigRedisAWSAuthWithRegion() testDataValidConfig { + c := testDataValidConfig{name: "Redis - AWS IAM auth with explicit region override"} + c.makeConfig = func(c *Config) { + c.Redis = RedisConfig{ + URL: newOptURLAbsoluteMustBeValid("rediss://my-cluster.amazonaws.com:6379"), + TLS: true, + Username: "iam-user", + AWSAuth: true, + AWSCacheName: "my-cache", + AWSRegion: "eu-west-1", + } + } + c.envVars = map[string]string{ + "USE_REDIS": "1", + "REDIS_URL": "rediss://my-cluster.amazonaws.com:6379", + "REDIS_TLS": "1", + "REDIS_USERNAME": "iam-user", + "REDIS_AWS_AUTH": "1", + "REDIS_AWS_CACHE_NAME": "my-cache", + "REDIS_AWS_REGION": "eu-west-1", + } + c.fileContent = ` +[Redis] +Url = "rediss://my-cluster.amazonaws.com:6379" +TLS = true +Username = "iam-user" +AWSAuth = true +AWSCacheName = "my-cache" +AWSRegion = "eu-west-1" +` + return c +} + +func makeValidConfigRedisAWSAuthServerless() testDataValidConfig { + c := testDataValidConfig{name: "Redis - AWS IAM auth for Serverless cache"} + c.makeConfig = func(c *Config) { + c.Redis = RedisConfig{ + URL: newOptURLAbsoluteMustBeValid("rediss://my-serverless.amazonaws.com:6379"), + TLS: true, + Username: "iam-user", + AWSAuth: true, + AWSCacheName: "my-serverless", + AWSServerless: true, + } + } + c.envVars = map[string]string{ + "USE_REDIS": "1", + "REDIS_URL": "rediss://my-serverless.amazonaws.com:6379", + "REDIS_TLS": "1", + "REDIS_USERNAME": "iam-user", + "REDIS_AWS_AUTH": "1", + "REDIS_AWS_CACHE_NAME": "my-serverless", + "REDIS_AWS_SERVERLESS": "1", + } + c.fileContent = ` +[Redis] +Url = "rediss://my-serverless.amazonaws.com:6379" +TLS = true +Username = "iam-user" +AWSAuth = true +AWSCacheName = "my-serverless" +AWSServerless = true +` + return c +} + +func makeValidConfigRedisAWSAuthLowercasesCacheName() testDataValidConfig { + c := testDataValidConfig{name: "Redis - AWS IAM auth lowercases cache name"} + c.makeConfig = func(c *Config) { + c.Redis = RedisConfig{ + URL: newOptURLAbsoluteMustBeValid("rediss://my-cluster.amazonaws.com:6379"), + TLS: true, + Username: "iam-user", + AWSAuth: true, + AWSCacheName: "my-cache", + } + } + c.envVars = map[string]string{ + "USE_REDIS": "1", + "REDIS_URL": "rediss://my-cluster.amazonaws.com:6379", + "REDIS_TLS": "1", + "REDIS_USERNAME": "iam-user", + "REDIS_AWS_AUTH": "1", + "REDIS_AWS_CACHE_NAME": "My-Cache", + } + c.fileContent = ` +[Redis] +Url = "rediss://my-cluster.amazonaws.com:6379" +TLS = true +Username = "iam-user" +AWSAuth = true +AWSCacheName = "My-Cache" +` + return c +} + func makeValidConfigConsulMinimal() testDataValidConfig { c := testDataValidConfig{name: "Consul - minimal parameters"} c.makeConfig = func(c *Config) { diff --git a/docs/configuration.md b/docs/configuration.md index ab1dc530..790c9d8b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -199,6 +199,10 @@ To learn more, read [Persistent storage](./persistent-storage.md). | `tls` | `REDIS_TLS` | Boolean | `false` | If `true`, will use a secure connection to Redis (not all Redis servers support this). If you specified a `redis://` URL, setting `tls` to `true` will change it to `rediss://`. | | `password` | `REDIS_PASSWORD` | String | | Optional password if Redis requires authentication. | | `username` | `REDIS_USERNAME` | String | | Optional username if Redis requires authentication. | +| `awsAuth` | `REDIS_AWS_AUTH` | Boolean | `false` | Enable AWS IAM authentication for ElastiCache. Requires `REDIS_TLS=true`, `REDIS_USERNAME`, and `REDIS_AWS_CACHE_NAME`. Mutually exclusive with `REDIS_PASSWORD`. | +| `awsCacheName` | `REDIS_AWS_CACHE_NAME` | String | | ElastiCache cluster name (lowercased automatically). Required when `REDIS_AWS_AUTH` is `true`. Used to construct the SigV4 presigned authentication URL. | +| `awsRegion` | `REDIS_AWS_REGION` | String | | AWS region override for SigV4 signing of ElastiCache IAM tokens. If empty, the region is resolved from the standard AWS credential chain (`AWS_DEFAULT_REGION`, `~/.aws/config`, IMDS, etc.). Only meaningful when `REDIS_AWS_AUTH` is `true`. | +| `awsServerless` | `REDIS_AWS_SERVERLESS` | Boolean | `false` | **Experimental — not functionally supported.** Adds `ResourceType=ServerlessCache` to the SigV4 presigned token URL, which is required for a Serverless cache to accept the IAM auth handshake. This covers authentication only; ElastiCache Serverless is cluster-mode-only and the Relay Proxy's Redis clients are not cluster-aware, so multi-key operations will fail with `CROSSSLOT`. Full Serverless support depends on cluster-aware client work that has not shipped. Do not use in production. See [Persistent storage](./persistent-storage.md#elasticache-serverless-experimental--not-functionally-supported). Only meaningful when `REDIS_AWS_AUTH` is `true`. | | `localTtl` | `CACHE_TTL` | Duration | `30s` | Length of time that database items can be cached in memory. | Note that the TLS and password options can also be specified as part of the URL: `rediss://` instead of `redis://` diff --git a/docs/persistent-storage.md b/docs/persistent-storage.md index 6a0ce1a3..d58fa314 100644 --- a/docs/persistent-storage.md +++ b/docs/persistent-storage.md @@ -47,6 +47,52 @@ If the database becomes unavailable, the Relay Proxy's behavior, based on its us The in-memory cache only helps SDKs using the Relay Proxy in proxy mode. SDKs configured to use daemon mode are connected to read directly from the database. To learn more, read [Configuring an SDK to use different modes](https://docs.launchdarkly.com/home/relay-proxy/using#configuring-an-sdk-to-use-different-modes). +## AWS IAM authentication for ElastiCache + +When running the Relay Proxy on AWS against an ElastiCache cluster, you can authenticate using an IAM identity instead of a static Redis password. This is most useful when Relay runs on EKS with IRSA or Pod Identity, since it removes the need to inject long-lived Redis credentials into the pod environment. Relay uses the pod's AWS credentials to generate a short-lived SigV4 presigned authentication token for each Redis connection. + +### AWS-side prerequisites + +- ElastiCache for Valkey 7.2+ or Redis OSS 7.0+. +- TLS enabled on the cache. +- An IAM-enabled ElastiCache user created with `authentication-mode Type=iam`, where the user's `user-id` and `username` are identical (AWS requires this for IAM auth). +- An IAM policy granting `elasticache:Connect` on both the cache resource and the user resource, attached to the role the Relay pod assumes. + +### Example configuration + +``` +USE_REDIS=1 +REDIS_HOST=my-cluster.abc123.use1.cache.amazonaws.com +REDIS_PORT=6379 +REDIS_TLS=true +REDIS_USERNAME=my-relay-user +REDIS_AWS_AUTH=true +REDIS_AWS_CACHE_NAME=my-cluster +``` + +The AWS region and credentials are picked up from the standard AWS environment (for example, the variables and files injected by IRSA or Pod Identity). To override the region used for signing, set `REDIS_AWS_REGION`: + +``` +REDIS_AWS_REGION=eu-west-1 +``` + +#### ElastiCache Serverless (experimental — not functionally supported) + +> [!WARNING] +> **`REDIS_AWS_SERVERLESS` is experimental and not yet functionally supported.** It currently covers only the IAM authentication handshake — when set, it adds `ResourceType=ServerlessCache` to the SigV4 presigned token, which is required for a Serverless cache to accept the connection at all. Without it, authentication against a Serverless cache fails with an opaque `WRONGPASS` error. +> +> However, authenticating successfully is not the same as working correctly. ElastiCache Serverless runs in **cluster mode only**, and the Relay Proxy's Redis clients are not cluster-aware (see the note above: "The Relay Proxy does not support clustered Redis or Redis Sentinel"). Multi-key operations — SDK data-store writes and the big-segments optimistic-locking transactions (`Watch`/`TxPipelined` across multiple keys) — will hit `CROSSSLOT` errors when those keys land in different hash slots, which is the normal case on a cluster. +> +> Full Serverless support is blocked on cluster-aware client work that **has not shipped**. This path has never been validated against a real Serverless cache. Do not rely on `REDIS_AWS_SERVERLESS` in production; use a non-serverless ElastiCache cache (single shard / node-based) with `REDIS_AWS_AUTH=true` instead. + +``` +REDIS_AWS_SERVERLESS=true +``` + +### Connection lifetime + +AWS enforces a 12-hour maximum lifetime on connections authenticated with IAM. The Relay Proxy transparently recycles its Redis connections before that limit is reached, so no operator action is required. IAM authentication tokens are generated automatically per Redis connection and are not cached across the pool. + ## DynamoDB storage limitation As described in the notes for the [LaunchDarkly Go SDK DynamoDB integration](https://github.com/launchdarkly/go-server-sdk-dynamodb/blob/master/README.md#data-size-limitation), which is the internal implementation used by the Relay Proxy, it is not possible to store more than 400KB of JSON data for any one feature flag or segment when using DynamoDB. diff --git a/go.mod b/go.mod index 513ea83d..c0526ec2 100644 --- a/go.mod +++ b/go.mod @@ -31,7 +31,7 @@ require ( github.com/launchdarkly/go-server-sdk-consul/v3 v3.0.1 github.com/launchdarkly/go-server-sdk-dynamodb/v4 v4.0.2 github.com/launchdarkly/go-server-sdk-evaluation/v4 v4.0.0 - github.com/launchdarkly/go-server-sdk-redis-redigo/v3 v3.0.3 + github.com/launchdarkly/go-server-sdk-redis-redigo/v3 v3.0.4-0.20260605234208-6b815581ee0f github.com/launchdarkly/go-server-sdk/v7 v7.15.1 github.com/launchdarkly/go-test-helpers/v3 v3.1.0 github.com/launchdarkly/opencensus-go-exporter-stackdriver v0.14.7 diff --git a/go.sum b/go.sum index 4a42c5a5..a4a3ea83 100644 --- a/go.sum +++ b/go.sum @@ -351,8 +351,8 @@ github.com/launchdarkly/go-server-sdk-dynamodb/v4 v4.0.2 h1:U/7GMLF2K5I290KkXTmn github.com/launchdarkly/go-server-sdk-dynamodb/v4 v4.0.2/go.mod h1:CY0DrFKo2eWqOVHKbA+ebDxgcE3S1qsA6WtQ9X8EVGg= github.com/launchdarkly/go-server-sdk-evaluation/v4 v4.0.0 h1:p66IaTRGgZ0fONBzZ6Ilvlg776nHwM/B4YGSIjFQ7qw= github.com/launchdarkly/go-server-sdk-evaluation/v4 v4.0.0/go.mod h1:vLWxdBnXoz3SbhbzAs3oZgr5NjW0XH2Oa+O1Ad6OygI= -github.com/launchdarkly/go-server-sdk-redis-redigo/v3 v3.0.3 h1:q2caIiyh1HFz9p/EwCp5CJWQ9iMkaCo6imPj0WgPKvY= -github.com/launchdarkly/go-server-sdk-redis-redigo/v3 v3.0.3/go.mod h1:QxE4Tn/RGlS8SeDp2tzQNPLXoHDWz4P86+uFYprmVZA= +github.com/launchdarkly/go-server-sdk-redis-redigo/v3 v3.0.4-0.20260605234208-6b815581ee0f h1:6dMOwyreKi8kv747KgxxPNgq+elwR4VESzUxWqp15xk= +github.com/launchdarkly/go-server-sdk-redis-redigo/v3 v3.0.4-0.20260605234208-6b815581ee0f/go.mod h1:QxE4Tn/RGlS8SeDp2tzQNPLXoHDWz4P86+uFYprmVZA= github.com/launchdarkly/go-server-sdk/v7 v7.15.1 h1:mGzt35HpzxoJbMiq4SAef3zknRdD3qQKMOiXjnCGhqM= github.com/launchdarkly/go-server-sdk/v7 v7.15.1/go.mod h1:ptjzQg8gjPhRhG0LkmVWnHMXuB3ZXyfQYMFUJn2rNzA= github.com/launchdarkly/go-test-helpers/v2 v2.3.2 h1:WX6qSzt7v8xz6d94nVcoil9ljuLTC/6OzQt0MhWYxsQ= diff --git a/internal/autoconfigcache/redis_store.go b/internal/autoconfigcache/redis_store.go index c30403d1..e30abf55 100644 --- a/internal/autoconfigcache/redis_store.go +++ b/internal/autoconfigcache/redis_store.go @@ -11,6 +11,7 @@ import ( "github.com/launchdarkly/go-sdk-common/v4/ldlog" "github.com/launchdarkly/ld-relay/v8/config" "github.com/launchdarkly/ld-relay/v8/internal/autoconfig" + "github.com/launchdarkly/ld-relay/v8/internal/awsredisauth" "github.com/launchdarkly/ld-relay/v8/internal/envfactory" ) @@ -48,6 +49,28 @@ func newRedisStore(redisConfig config.RedisConfig, cacheKey string, encKey []byt MinVersion: tls.VersionTLS12, } } + + if redisConfig.AWSAuth { + // Clear any password/username that came in via URL parsing. Otherwise + // go-redis runs a pipeline-AUTH before OnConnect fires and sends the + // static credentials to ElastiCache, which rejects them. + uo.Username = "" + uo.Password = "" + + provider, provErr := awsredisauth.SharedTokenProvider(context.Background(), redisConfig, loggers) + if provErr != nil { + return nil, provErr + } + uo.OnConnect = func(ctx context.Context, cn *redis.Conn) error { + tok, err := provider.Token(ctx) + if err != nil { + return err + } + return cn.AuthACL(ctx, redisConfig.Username, tok).Err() + } + uo.MaxConnAge = awsredisauth.JitteredMaxConnAge() + } + client := redis.NewUniversalClient(uo) ctx, cancel := context.WithCancel(context.Background()) return &redisStore{client: client, ctx: ctx, cancel: cancel, hashKey: cacheKey, encKey: encKey, loggers: loggers}, nil diff --git a/internal/autoconfigcache/redis_store_aws_test.go b/internal/autoconfigcache/redis_store_aws_test.go new file mode 100644 index 00000000..558374b2 --- /dev/null +++ b/internal/autoconfigcache/redis_store_aws_test.go @@ -0,0 +1,84 @@ +package autoconfigcache + +import ( + "context" + "errors" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/launchdarkly/go-configtypes" + "github.com/launchdarkly/go-sdk-common/v4/ldlog" + "github.com/launchdarkly/ld-relay/v8/config" + "github.com/launchdarkly/ld-relay/v8/internal/awsredisauth" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// errCredsProvider is a test-only aws.CredentialsProvider that always returns an error. +type errCredsProvider struct { + err error +} + +func (e *errCredsProvider) Retrieve(_ context.Context) (aws.Credentials, error) { + return aws.Credentials{}, e.err +} + +// staticAWSConfig returns an aws.Config with deterministic static credentials for tests. +func staticAWSConfig() aws.Config { + return aws.Config{ + Region: "us-east-1", + Credentials: credentials.NewStaticCredentialsProvider( + "AKIAIOSFODNN7EXAMPLE", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "", + ), + } +} + +// awsRedisConfig returns a RedisConfig with AWSAuth=true and all required companion fields. +func awsRedisConfig() config.RedisConfig { + rc := config.RedisConfig{} + rc.URL, _ = configtypes.NewOptURLAbsoluteFromString("rediss://my-cache.abc123.use1.cache.amazonaws.com:6379") + rc.AWSAuth = true + rc.AWSCacheName = "my-cache" + rc.Username = "iam-user-01" + rc.TLS = true + return rc +} + +// TestNewRedisStore_AWSAuthCredentialsError verifies that when AWSAuth=true +// and the AWS credentials provider returns an error, NewTokenProviderFromAWSConfig +// returns that error at construction time (fail-fast token verification). +func TestNewRedisStore_AWSAuthCredentialsError(t *testing.T) { + sentinelErr := errors.New("no credentials available") + + cfg := aws.Config{ + Region: "us-east-1", + Credentials: &errCredsProvider{err: sentinelErr}, + } + rc := awsRedisConfig() + + _, err := awsredisauth.NewTokenProviderFromAWSConfig(context.Background(), cfg, rc) + require.Error(t, err) + assert.ErrorIs(t, err, sentinelErr) +} + +// TestNewRedisStore_AWSAuthSuccess verifies that a valid aws.Config produces a usable provider. +func TestNewRedisStore_AWSAuthSuccess(t *testing.T) { + provider, err := awsredisauth.NewTokenProviderFromAWSConfig(context.Background(), staticAWSConfig(), awsRedisConfig()) + require.NoError(t, err) + require.NotNil(t, provider) +} + +// TestNewRedisStore_AWSAuth_ErrorPropagated verifies that newRedisStore returns an error +// when AWSAuth=true and no real AWS credentials are available (CI environment). +func TestNewRedisStore_AWSAuth_ErrorPropagated(t *testing.T) { + awsredisauth.ResetSharedTokenProvidersForTest() + defer awsredisauth.ResetSharedTokenProvidersForTest() + + rc := awsRedisConfig() + _, err := newRedisStore(rc, "test-cache-key", make([]byte, 32), ldlog.NewDisabledLoggers()) + // In CI (no AWS credentials or region), startup must fail fast. + assert.Error(t, err) +} diff --git a/internal/awsredisauth/integration_test.go b/internal/awsredisauth/integration_test.go new file mode 100644 index 00000000..cb2603b5 --- /dev/null +++ b/internal/awsredisauth/integration_test.go @@ -0,0 +1,511 @@ +//go:build integration + +package awsredisauth_test + +// Layer 2 integration test: mock TTL-enforcing Redis-protocol server. +// +// Verifies that the redigo PasswordProvider path correctly re-authenticates on +// reconnect when tokens expire. The mock server implements real AWS semantics: +// +// - Parses the SigV4 presigned token supplied in AUTH to extract X-Amz-Date +// and X-Amz-Expires. +// - Rejects AUTH with an already-expired token (closing the connection). This +// matches the AWS doc: "If the connection is re-authenticated with an expired +// token, the authentication request will be rejected." +// - Once authenticated, the connection remains alive for subsequent commands +// regardless of token expiry (AUTH-once semantics). +// - Responds to PING, MULTI, EXEC, WATCH, UNWATCH, SET, DEL, HSET, HGET, +// HGETALL, EXISTS, and SELECT with minimal but correct RESP replies. +// +// The test uses MaxConnLifetime = tokenLifetime (5s) to force the pool to recycle +// connections at each expiry boundary, proving the PasswordProvider is invoked +// with a fresh token on each reconnect. +// +// Run with: +// +// go test -v -tags integration -run TestTokenRotation -timeout 60s ./internal/awsredisauth/ + +import ( + "bufio" + "context" + "fmt" + "net" + "net/url" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + ldredis "github.com/launchdarkly/go-server-sdk-redis-redigo/v3" + redigo "github.com/gomodule/redigo/redis" + "github.com/launchdarkly/ld-relay/v8/internal/awsredisauth" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// --------------------------------------------------------------------------- +// Mock RESP server +// --------------------------------------------------------------------------- + +// mockServer is a minimal Redis-protocol server that enforces token TTLs. +type mockServer struct { + listener net.Listener + addr string + + // counters (atomic) + connectionsAccepted int64 + noauthRejections int64 + commandsServed int64 +} + +func newMockServer(t *testing.T) *mockServer { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + s := &mockServer{ + listener: ln, + addr: ln.Addr().String(), + } + go s.serve() + t.Cleanup(func() { _ = ln.Close() }) + return s +} + +func (s *mockServer) serve() { + for { + conn, err := s.listener.Accept() + if err != nil { + return // listener closed + } + atomic.AddInt64(&s.connectionsAccepted, 1) + go s.handleConn(conn) + } +} + +// connState holds per-connection auth expiry info. +type connState struct { + authed bool + expiresAt time.Time // zero means not yet authed +} + +func (s *mockServer) handleConn(c net.Conn) { + defer c.Close() //nolint:errcheck + + r := bufio.NewReader(c) + state := &connState{} + + for { + cmd, args, err := readCommand(r) + if err != nil { + return // client disconnected + } + + reply, closeConn := s.dispatch(cmd, args, state) + _, _ = fmt.Fprint(c, reply) + if closeConn { + return + } + } +} + +// dispatch handles a single RESP command and returns the reply string plus +// whether the server should close the connection after replying. +func (s *mockServer) dispatch(cmd string, args []string, state *connState) (reply string, closeConn bool) { + upper := strings.ToUpper(cmd) + + switch upper { + case "AUTH": + // AWS ElastiCache IAM auth requires the ACL form: AUTH . + // Single-arg AUTH is the legacy form and is rejected here to mirror + // real ElastiCache behavior — and to ensure the test would fail if + // the production wiring ever stops sending DialUsername alongside + // PasswordProvider. + if len(args) != 2 { + return "-ERR ElastiCache IAM auth requires the two-argument form AUTH \r\n", false + } + token := args[1] + + expiry, err := parseTokenExpiry(token) + if err != nil { + return "-ERR invalid token: " + err.Error() + "\r\n", false + } + + // Real AWS semantics: reject AUTH with an already-expired token. + // (Per docs: "If the connection is re-authenticated with an expired token, + // the authentication request will be rejected.") + if time.Now().After(expiry) { + atomic.AddInt64(&s.noauthRejections, 1) + return "-NOAUTH Authentication failed: token has expired\r\n", true + } + + state.authed = true + state.expiresAt = expiry + atomic.AddInt64(&s.commandsServed, 1) + return "+OK\r\n", false + + case "PING": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + if len(args) == 0 { + return "+PONG\r\n", false + } + // PING -> bulk string echo + msg := args[0] + return fmt.Sprintf("$%d\r\n%s\r\n", len(msg), msg), false + + case "SELECT": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + return "+OK\r\n", false + + case "MULTI": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + return "+OK\r\n", false + + case "EXEC": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + // Return an empty array — no queued commands in our simplified model. + return "*0\r\n", false + + case "WATCH", "UNWATCH": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + return "+OK\r\n", false + + case "DEL": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + return ":0\r\n", false + + case "SET": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + return "+OK\r\n", false + + case "HSET": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + return ":0\r\n", false + + case "HGET": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + // Return nil bulk string (key not found). + return "$-1\r\n", false + + case "HGETALL": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + // Return empty array. + return "*0\r\n", false + + case "EXISTS": + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + atomic.AddInt64(&s.commandsServed, 1) + // Return 0 — key does not exist. + return ":0\r\n", false + + default: + if !state.authed { + return "-NOAUTH Authentication required\r\n", false + } + // Gracefully handle any unlisted command so the test doesn't stall. + atomic.AddInt64(&s.commandsServed, 1) + return "+OK\r\n", false + } +} + +// --------------------------------------------------------------------------- +// RESP parser +// --------------------------------------------------------------------------- + +// readCommand reads one RESP command from the reader. Supports both inline +// commands and the array-of-bulk-strings form (*N\r\n$Len\r\n...). +func readCommand(r *bufio.Reader) (cmd string, args []string, err error) { + line, err := readLine(r) + if err != nil { + return "", nil, err + } + + if strings.HasPrefix(line, "*") { + // Array form: * + n, err := strconv.Atoi(line[1:]) + if err != nil || n < 1 { + return "", nil, fmt.Errorf("invalid array count: %s", line) + } + parts := make([]string, 0, n) + for i := 0; i < n; i++ { + s, err := readBulkString(r) + if err != nil { + return "", nil, err + } + parts = append(parts, s) + } + return parts[0], parts[1:], nil + } + + // Inline command (e.g. "PING\r\n" or "AUTH password\r\n") + parts := strings.Fields(line) + if len(parts) == 0 { + return readCommand(r) // skip blank lines + } + return parts[0], parts[1:], nil +} + +func readLine(r *bufio.Reader) (string, error) { + line, err := r.ReadString('\n') + if err != nil { + return "", err + } + return strings.TrimRight(line, "\r\n"), nil +} + +func readBulkString(r *bufio.Reader) (string, error) { + line, err := readLine(r) + if err != nil { + return "", err + } + if !strings.HasPrefix(line, "$") { + return "", fmt.Errorf("expected bulk string, got: %s", line) + } + n, err := strconv.Atoi(line[1:]) + if err != nil { + return "", fmt.Errorf("invalid bulk string length: %s", line) + } + if n < 0 { + return "", nil // null bulk string + } + buf := make([]byte, n+2) // +2 for \r\n + if _, err := r.Read(buf); err != nil { + return "", err + } + return string(buf[:n]), nil +} + +// --------------------------------------------------------------------------- +// Token expiry parser +// --------------------------------------------------------------------------- + +// parseTokenExpiry parses a SigV4 presigned token (scheme-stripped URL) and +// returns the absolute expiry time derived from X-Amz-Date + X-Amz-Expires. +// +// The token is the scheme-stripped form, e.g.: +// +// my-cache/?Action=connect&User=u&X-Amz-Date=20240101T000000Z&X-Amz-Expires=5&... +// +// net/url.Parse can't handle a URL that starts with a hostname (no scheme), +// so we prepend "https://" before parsing. +func parseTokenExpiry(token string) (time.Time, error) { + if !strings.Contains(token, "://") { + token = "https://" + token + } + u, err := url.Parse(token) + if err != nil { + return time.Time{}, fmt.Errorf("parsing token URL: %w", err) + } + q := u.Query() + + dateStr := q.Get("X-Amz-Date") + if dateStr == "" { + return time.Time{}, fmt.Errorf("token missing X-Amz-Date") + } + authTime, err := time.Parse("20060102T150405Z", dateStr) + if err != nil { + return time.Time{}, fmt.Errorf("parsing X-Amz-Date %q: %w", dateStr, err) + } + + expiresStr := q.Get("X-Amz-Expires") + if expiresStr == "" { + return time.Time{}, fmt.Errorf("token missing X-Amz-Expires") + } + expiresSeconds, err := strconv.ParseInt(expiresStr, 10, 64) + if err != nil { + return time.Time{}, fmt.Errorf("parsing X-Amz-Expires %q: %w", expiresStr, err) + } + + return authTime.Add(time.Duration(expiresSeconds) * time.Second), nil +} + +// --------------------------------------------------------------------------- +// Test +// --------------------------------------------------------------------------- + +// TestTokenRotation is the Layer 2 integration test. It drives the redigo +// data store path against the mock TTL-enforcing server for ~20 seconds +// (crossing 3-4 x 5-second expiry boundaries) and asserts that every +// command succeeds. +// +// Pass: zero command failures (the pool reconnects with a fresh token on each +// +// expiry boundary before the command is retried by the caller). +// +// Run: go test -v -tags integration -run TestTokenRotation -timeout 60s ./internal/awsredisauth/ +func TestTokenRotation(t *testing.T) { + const ( + tokenLifetime = 5 * time.Second + testDuration = 20 * time.Second + opInterval = 100 * time.Millisecond + ) + + // 1. Start the mock server. + mock := newMockServer(t) + + // 2. Build a TokenProvider with a 5-second lifetime. + awsCfg := aws.Config{ + Region: "us-east-1", + Credentials: credentials.NewStaticCredentialsProvider( + "AKIAIOSFODNN7EXAMPLE", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "", + ), + } + provider, err := awsredisauth.NewTokenProvider(awsCfg, "test-cache", "iam-user-01", + awsredisauth.Options{TokenLifetime: tokenLifetime}) + require.NoError(t, err) + + // 3. Build the redigo pool using the same builder path as makeRedisDataStoreBuilder. + // We use the StoreBuilder internals directly to stay close to production code + // without needing to wire up the full relay SDK context. + // + // The pool mirrors the production config: PasswordProvider + MaxConnLifetime. + redisURL := "redis://" + mock.addr + pool := newRedigoPool(redisURL, "iam-user-01", provider.Token, tokenLifetime) + defer pool.Close() //nolint:errcheck + + // 4. Drive traffic for testDuration. + deadline := time.Now().Add(testDuration) + var failures int64 + var ops int64 + + for time.Now().Before(deadline) { + c := pool.Get() + connErr := c.Err() + if connErr == nil { + // Issue a lightweight command. + _, cmdErr := c.Do("PING") + if cmdErr != nil { + // A real NOAUTH error here means the pool handed us a stale + // authenticated connection whose token expired and the server + // closed it. Redigo should have detected this via TestOnBorrow + // (which does a PING) and re-dialed. If we see this it means + // the rotation failed. + t.Logf("command error at %s: %v", time.Now().Format(time.RFC3339), cmdErr) + atomic.AddInt64(&failures, 1) + } + atomic.AddInt64(&ops, 1) + } else { + // Pool dial failed entirely. + t.Logf("pool.Get error at %s: %v", time.Now().Format(time.RFC3339), connErr) + atomic.AddInt64(&failures, 1) + } + _ = c.Close() + time.Sleep(opInterval) + } + + total := atomic.LoadInt64(&ops) + fail := atomic.LoadInt64(&failures) + conns := atomic.LoadInt64(&mock.connectionsAccepted) + noauth := atomic.LoadInt64(&mock.noauthRejections) + served := atomic.LoadInt64(&mock.commandsServed) + + t.Logf("Test complete: %d ops, %d failures, %d connections, %d NOAUTH rejections, %d commands served", + total, fail, conns, noauth, served) + + // Expiry boundaries crossed: testDuration / tokenLifetime = ~4. + // Each boundary triggers a MaxConnLifetime-driven pool recycle + fresh dial. + expectedBoundaries := int64(testDuration / tokenLifetime) + t.Logf("Expected ~%d expiry boundaries (pool recycles); mock accepted %d connections", + expectedBoundaries, conns) + + // Primary assertion: no command visible to the caller failed. + // AUTH-once semantics: established connections stay alive regardless of token + // expiry. The pool recycles connections via MaxConnLifetime and re-authenticates + // with a fresh token each time. All caller-visible commands succeed. + assert.Equal(t, int64(0), fail, "expected zero command failures; the pool should reconnect transparently on token expiry") + + // Sanity: we issued a reasonable number of operations. + assert.Greater(t, total, int64(0), "at least one operation must have been issued") + + // Sanity: pool accepted multiple connections (reconnects happened at each boundary). + // With 5s MaxConnLifetime and 20s test duration, we expect ~4-5 connections. + assert.Greater(t, conns, int64(1), "mock should have accepted more than one connection (pool recycles on MaxConnLifetime)") + + // Sanity: no NOAUTH rejections (every token provided was fresh at AUTH time). + // A non-zero count here would indicate the TokenProvider returned a stale token. + assert.Equal(t, int64(0), noauth, "TokenProvider should always return a fresh (non-expired) token at AUTH time") + + // Sanity: the mock served a reasonable number of commands. + assert.Greater(t, served, int64(0), "mock should have served commands") +} + +// newRedigoPool creates a redigo pool wired with the given PasswordProvider, +// mirroring the production configuration in internal/sdks/data_stores.go's +// makeRedisDataStoreBuilder (PasswordProvider + MaxConnLifetime). We use the +// ldredis.StoreBuilder to exercise the exact production pool-construction path. +// +// However StoreBuilder.Build requires a subsystems.ClientContext. To avoid +// pulling in the full relay SDK plumbing for this test, we construct the pool +// directly via the public API exposed on StoreBuilder — which conveniently +// accepts just a URL + provider — by calling the internal newPool path via the +// ldredis package's exported helpers. +// +// Actually, newPool is unexported. We use the simpler approach: construct a +// raw redigo.Pool directly with the same parameters that StoreBuilder would use. +// This is intentional: the mock drives the *connection+auth* path, and the pool +// parameters (MaxIdle, MaxActive, TestOnBorrow, PasswordProvider closure) are +// what we need to replicate. +func newRedigoPool(redisURL, username string, tokenFn func(ctx context.Context) (string, error), maxLifetime time.Duration) *redigo.Pool { + return &redigo.Pool{ + MaxIdle: 5, + MaxActive: 5, + Wait: true, + IdleTimeout: 30 * time.Second, + MaxConnLifetime: maxLifetime, + Dial: func() (redigo.Conn, error) { + pw, err := tokenFn(context.Background()) + if err != nil { + return nil, err + } + // Matches the production wiring in makeRedisDataStoreBuilder: + // DialUsername is required so redigo emits AUTH + // (the ACL form ElastiCache IAM auth demands). + return redigo.DialURL(redisURL, redigo.DialUsername(username), redigo.DialPassword(pw)) + }, + TestOnBorrow: func(c redigo.Conn, t time.Time) error { + _, err := c.Do("PING") + return err + }, + } +} + +// Compile-time check: ldredis.StoreBuilder is used in the import so it stays +// reachable and any API changes surface at build time. +var _ = ldredis.DataStore diff --git a/internal/awsredisauth/jitter.go b/internal/awsredisauth/jitter.go new file mode 100644 index 00000000..b5106a8c --- /dev/null +++ b/internal/awsredisauth/jitter.go @@ -0,0 +1,56 @@ +package awsredisauth + +import ( + "math/rand/v2" + "time" +) + +const ( + // maxConnAgeBase is the target maximum connection age for ElastiCache IAM-auth + // pools. AWS enforces a hard 12-hour ceiling on IAM-authenticated connections, + // so we recycle well before that limit. + maxConnAgeBase = 11 * time.Hour + + // maxConnAgeJitter is the maximum amount subtracted from maxConnAgeBase to + // produce a per-pool jittered value. This staggers reconnects across pods and + // across the three Relay pools within a single pod. The effective range is + // [maxConnAgeBase - maxConnAgeJitter, maxConnAgeBase] = [10h30m, 11h]. + maxConnAgeJitter = 30 * time.Minute +) + +// JitteredMaxConnAge returns a uniformly random connection max-age in the range +// [10h30m, 11h]. Callers should invoke it once per pool construction; each pool +// may (and should) get its own independent value. +// +// # Why jitter? +// +// All three Relay Redis pools use IAM authentication, which means every connection +// must be recycled before the AWS-enforced 12-hour IAM token lifetime. Without +// jitter, every connection in every pool would expire at the same wall-clock time, +// causing a thundering-herd reconnect storm against ElastiCache. +// +// This per-pool jitter staggers reconnects in two dimensions: +// +// - Across pods: different pods draw independent random values, so their recycle +// windows are spread across the 30-minute jitter band. +// - Across pools within a pod: the three pools (SDK data store, big-segments store, +// auto-config cache) each call JitteredMaxConnAge independently and will almost +// always draw different values. +// +// # Residual thundering-herd (acknowledged limitation) +// +// This single-per-pool duration does NOT stagger connections within a single pool +// that were created at the same time. For example, if a Redis failover triggers +// simultaneous reconnects of all idle connections in the pool, they will all receive +// the same MaxConnAge value and therefore all expire together ~11h later, causing +// another burst at that time. +// +// True within-pool staggering requires per-connection max-age tracking, which +// neither go-redis/v8 nor redigo supports via a single duration. This residual is +// resolved by the planned go-redis v9 migration, which provides a native +// CredentialsProvider interface that re-auths in place on each connection refresh +// instead of forcing a full TCP recycle. +func JitteredMaxConnAge() time.Duration { + jitter := time.Duration(rand.Int64N(int64(maxConnAgeJitter))) //nolint:gosec // connection-age jitter is not security-sensitive; a weak RNG is fine + return maxConnAgeBase - jitter +} diff --git a/internal/awsredisauth/jitter_test.go b/internal/awsredisauth/jitter_test.go new file mode 100644 index 00000000..b2271db6 --- /dev/null +++ b/internal/awsredisauth/jitter_test.go @@ -0,0 +1,38 @@ +package awsredisauth + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +// TestJitteredMaxConnAge_InRange verifies that JitteredMaxConnAge always returns a +// duration in the expected range [10h30m, 11h]. +func TestJitteredMaxConnAge_InRange(t *testing.T) { + const iterations = 10_000 + const lower = maxConnAgeBase - maxConnAgeJitter // 10h30m + const upper = maxConnAgeBase // 11h + + for i := 0; i < iterations; i++ { + d := JitteredMaxConnAge() + assert.GreaterOrEqualf(t, d, lower, + "iteration %d: got %v, want >= %v", i, d, lower) + assert.LessOrEqualf(t, d, upper, + "iteration %d: got %v, want <= %v", i, d, upper) + } +} + +// TestJitteredMaxConnAge_Spread verifies that consecutive calls produce some variation, +// confirming the jitter is actually random. With 1000 calls and a 30-minute range in +// nanoseconds (~1.8e12), the probability that all values are identical is astronomically small. +func TestJitteredMaxConnAge_Spread(t *testing.T) { + const iterations = 1000 + seen := make(map[time.Duration]struct{}, iterations) + for i := 0; i < iterations; i++ { + seen[JitteredMaxConnAge()] = struct{}{} + } + // With 1000 draws from a ~1.8e12-value range, expect many distinct values. + // Requiring at least 2 is a minimal sanity check. + assert.Greater(t, len(seen), 1, "JitteredMaxConnAge should produce multiple distinct values") +} diff --git a/internal/awsredisauth/provider_from_config.go b/internal/awsredisauth/provider_from_config.go new file mode 100644 index 00000000..6ef62f6f --- /dev/null +++ b/internal/awsredisauth/provider_from_config.go @@ -0,0 +1,50 @@ +package awsredisauth + +import ( + "context" + "fmt" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/launchdarkly/ld-relay/v8/config" +) + +// NewTokenProviderFromRedisConfig constructs a TokenProvider for the given RedisConfig. +// It loads AWS credentials via the SDK default chain (environment variables, ~/.aws/credentials, +// IRSA web-identity token, EKS Pod Identity, etc.), then performs a fail-fast verification +// by calling Token() once. Any error — missing credentials, empty region, or network failure +// during the initial STS call — is returned immediately so relay startup fails with a clear +// message rather than silently deferring the failure to the first Redis connection. +// +// This is the recommended entry point for all relay wiring sites. Callers must only invoke +// this function when redisConfig.AWSAuth is true. +func NewTokenProviderFromRedisConfig(ctx context.Context, redisConfig config.RedisConfig) (TokenProvider, error) { + cfg, err := awsconfig.LoadDefaultConfig(ctx) + if err != nil { + return nil, fmt.Errorf("awsredisauth: loading AWS config: %w", err) + } + return NewTokenProviderFromAWSConfig(ctx, cfg, redisConfig) +} + +// NewTokenProviderFromAWSConfig is like NewTokenProviderFromRedisConfig but accepts a +// pre-constructed aws.Config rather than loading one from the default chain. This is the +// testable entry point used by unit tests that need to inject a controlled aws.Config +// (e.g., a credentials provider that returns an error). +func NewTokenProviderFromAWSConfig(ctx context.Context, cfg aws.Config, redisConfig config.RedisConfig) (TokenProvider, error) { + opts := Options{ + Region: redisConfig.AWSRegion, + Serverless: redisConfig.AWSServerless, + } + provider, err := NewTokenProvider(cfg, redisConfig.AWSCacheName, redisConfig.Username, opts) + if err != nil { + return nil, err + } + + // Fail-fast verification: call Token() once at startup. This surfaces misconfigured + // credentials or a missing AWS region before any Redis connection is attempted. + if _, err := provider.Token(ctx); err != nil { + return nil, fmt.Errorf("awsredisauth: startup token verification failed: %w", err) + } + + return provider, nil +} diff --git a/internal/awsredisauth/provider_from_config_test.go b/internal/awsredisauth/provider_from_config_test.go new file mode 100644 index 00000000..fc60446f --- /dev/null +++ b/internal/awsredisauth/provider_from_config_test.go @@ -0,0 +1,111 @@ +package awsredisauth + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/launchdarkly/ld-relay/v8/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// makeTestRedisConfig returns a RedisConfig with AWSAuth=true and the required companion +// fields populated. The URL uses a dummy host; no actual connection is made in these tests. +func makeTestRedisConfig() config.RedisConfig { + rc := config.RedisConfig{} + rc.AWSAuth = true + rc.AWSCacheName = "my-cache" + rc.Username = "iam-user-01" + return rc +} + +// TestNewTokenProviderFromAWSConfig_CredentialsErrorPropagates verifies that when the +// aws.Config has a credentials provider that fails Retrieve, NewTokenProviderFromAWSConfig +// returns an error at construction time (fail-fast verification). +func TestNewTokenProviderFromAWSConfig_CredentialsErrorPropagates(t *testing.T) { + sentinelErr := errors.New("no credentials available") + + cfg := aws.Config{ + Region: "us-east-1", + Credentials: &errCredsProvider{err: sentinelErr}, + } + + _, err := NewTokenProviderFromAWSConfig(context.Background(), cfg, makeTestRedisConfig()) + require.Error(t, err) + assert.ErrorIs(t, err, sentinelErr, "credentials error must be wrapped and propagated") +} + +// TestNewTokenProviderFromAWSConfig_SuccessWithStaticCreds verifies that a valid aws.Config +// produces a usable TokenProvider. +func TestNewTokenProviderFromAWSConfig_SuccessWithStaticCreds(t *testing.T) { + provider, err := NewTokenProviderFromAWSConfig(context.Background(), staticCreds(), makeTestRedisConfig()) + require.NoError(t, err) + require.NotNil(t, provider) + + // Provider should be usable immediately. + tok, err := provider.Token(context.Background()) + require.NoError(t, err) + assert.NotEmpty(t, tok) +} + +// TestNewTokenProviderFromAWSConfig_RegionOverride verifies that AWSRegion in +// RedisConfig is threaded through to the SigV4 signing region. +func TestNewTokenProviderFromAWSConfig_RegionOverride(t *testing.T) { + const overrideRegion = "ap-northeast-1" + + rc := makeTestRedisConfig() + rc.AWSRegion = overrideRegion + + // Provide a cfg with a different region to confirm override wins. + cfg := aws.Config{ + Region: "us-east-1", + Credentials: staticCreds().Credentials, + } + + provider, err := NewTokenProviderFromAWSConfig(context.Background(), cfg, rc) + require.NoError(t, err) + + tok, err := provider.Token(context.Background()) + require.NoError(t, err) + + u := parseToken(t, tok) + credential := u.Query().Get("X-Amz-Credential") + assert.True(t, + strings.Contains(credential, "/"+overrideRegion+"/elasticache/aws4_request"), + "token credential %q must use AWSRegion override %q", credential, overrideRegion, + ) +} + +// TestNewTokenProviderFromAWSConfig_ServerlessFlag verifies that AWSServerless=true in +// RedisConfig causes ResourceType=ServerlessCache in the generated token. +func TestNewTokenProviderFromAWSConfig_ServerlessFlag(t *testing.T) { + rc := makeTestRedisConfig() + rc.AWSServerless = true + + provider, err := NewTokenProviderFromAWSConfig(context.Background(), staticCreds(), rc) + require.NoError(t, err) + + tok, err := provider.Token(context.Background()) + require.NoError(t, err) + + u := parseToken(t, tok) + assert.Equal(t, "ServerlessCache", u.Query().Get("ResourceType"), + "AWSServerless=true must produce ResourceType=ServerlessCache in token") +} + +// TestNewTokenProviderFromAWSConfig_ServerlessAbsent verifies that when AWSServerless +// is false (default), no ResourceType parameter appears in the token. +func TestNewTokenProviderFromAWSConfig_ServerlessAbsent(t *testing.T) { + provider, err := NewTokenProviderFromAWSConfig(context.Background(), staticCreds(), makeTestRedisConfig()) + require.NoError(t, err) + + tok, err := provider.Token(context.Background()) + require.NoError(t, err) + + u := parseToken(t, tok) + assert.Empty(t, u.Query().Get("ResourceType"), + "AWSServerless=false must not include ResourceType in token") +} diff --git a/internal/awsredisauth/shared_provider.go b/internal/awsredisauth/shared_provider.go new file mode 100644 index 00000000..5cafe6c9 --- /dev/null +++ b/internal/awsredisauth/shared_provider.go @@ -0,0 +1,109 @@ +package awsredisauth + +import ( + "context" + "fmt" + "sync" + + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/launchdarkly/go-sdk-common/v4/ldlog" + "github.com/launchdarkly/ld-relay/v8/config" +) + +// providerCacheKey is the fingerprint used to deduplicate SharedTokenProvider calls. +// Fields that affect the signed token or credential-loading behaviour are included; +// fields that are irrelevant to the provider itself (e.g. host/port/TLS) are omitted. +type providerCacheKey struct { + cacheName string + username string + region string + serverless bool +} + +// sharedProviderEntry is a cached provider plus its resolved region (for logging). +type sharedProviderEntry struct { + provider TokenProvider + resolvedRegion string +} + +// Process-wide memoization cache: one TokenProvider (and its AWS credential +// cache) shared across the three Redis wiring sites. Package-global by design. +var ( + sharedProvidersMu sync.Mutex //nolint:gochecknoglobals + sharedProviders = map[providerCacheKey]sharedProviderEntry{} //nolint:gochecknoglobals +) + +// ResetSharedTokenProvidersForTest clears the shared provider cache. It must only be +// called from tests; calling it in production code causes data races and re-probes. +func ResetSharedTokenProvidersForTest() { + sharedProvidersMu.Lock() + sharedProviders = map[providerCacheKey]sharedProviderEntry{} + sharedProvidersMu.Unlock() +} + +// SharedTokenProvider returns a memoized TokenProvider for the given RedisConfig. +// On the first call for a given (cacheName, username, region, serverless) fingerprint +// it loads the AWS default config, constructs a TokenProvider, performs a fail-fast +// Token() probe, and logs a startup info line. Subsequent calls with the same +// fingerprint return the cached instance without re-probing or re-logging. +// +// The function is concurrency-safe. +// +// Callers must only invoke this function when redisConfig.AWSAuth is true. +func SharedTokenProvider(ctx context.Context, redisConfig config.RedisConfig, loggers ldlog.Loggers) (TokenProvider, error) { + key := providerCacheKey{ + cacheName: redisConfig.AWSCacheName, + username: redisConfig.Username, + region: redisConfig.AWSRegion, + serverless: redisConfig.AWSServerless, + } + + sharedProvidersMu.Lock() + defer sharedProvidersMu.Unlock() + + if entry, ok := sharedProviders[key]; ok { + return entry.provider, nil + } + + // Not cached yet — build a new provider. + cfg, err := awsconfig.LoadDefaultConfig(ctx) + if err != nil { + return nil, fmt.Errorf("awsredisauth: loading AWS config: %w", err) + } + + // Apply region override if specified. + if redisConfig.AWSRegion != "" { + cfg.Region = redisConfig.AWSRegion + } + + opts := Options{ + Region: redisConfig.AWSRegion, + Serverless: redisConfig.AWSServerless, + } + + provider, err := NewTokenProvider(cfg, redisConfig.AWSCacheName, redisConfig.Username, opts) + if err != nil { + return nil, err + } + + // Fail-fast probe: call Token() once at startup. This surfaces misconfigured + // credentials or a missing/unresolvable AWS region before any Redis connection + // is attempted. + if _, err := provider.Token(ctx); err != nil { + return nil, fmt.Errorf("awsredisauth: startup token verification failed: %w", err) + } + + // Resolve the effective region for the log line. Options.Region takes precedence; + // fall back to whatever the SDK resolved from the credential chain. + resolvedRegion := provider.(*tokenProvider).region //nolint:forcetypeassert + + loggers.Infof("ElastiCache IAM auth enabled (cache=%s, region=%s, serverless=%t)", + redisConfig.AWSCacheName, resolvedRegion, redisConfig.AWSServerless) + + entry := sharedProviderEntry{ + provider: provider, + resolvedRegion: resolvedRegion, + } + sharedProviders[key] = entry + return provider, nil +} diff --git a/internal/awsredisauth/shared_provider_test.go b/internal/awsredisauth/shared_provider_test.go new file mode 100644 index 00000000..0ed5b103 --- /dev/null +++ b/internal/awsredisauth/shared_provider_test.go @@ -0,0 +1,186 @@ +package awsredisauth + +import ( + "context" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/launchdarkly/go-sdk-common/v4/ldlog" + "github.com/launchdarkly/ld-relay/v8/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// sharedProviderTestConfig constructs a test aws.Config that LoadDefaultConfig would +// normally produce, so that NewTokenProvider succeeds. SharedTokenProvider calls +// LoadDefaultConfig internally; we can't inject a cfg here like we can in +// NewTokenProviderFromAWSConfig. Instead we rely on the internal test bypass approach: +// we expose a separate sharedTokenProviderWithConfig path for tests. + +// sharedTokenProviderWithConfig is like SharedTokenProvider but accepts a pre-built +// aws.Config so tests can avoid LoadDefaultConfig. Declared here and implemented +// alongside SharedTokenProvider for testability without changing the public API. +func sharedTokenProviderWithConfig( + ctx context.Context, + cfg aws.Config, + redisConfig config.RedisConfig, + loggers ldlog.Loggers, +) (TokenProvider, error) { + key := providerCacheKey{ + cacheName: redisConfig.AWSCacheName, + username: redisConfig.Username, + region: redisConfig.AWSRegion, + serverless: redisConfig.AWSServerless, + } + + sharedProvidersMu.Lock() + defer sharedProvidersMu.Unlock() + + if entry, ok := sharedProviders[key]; ok { + return entry.provider, nil + } + + // Apply region override if specified. + if redisConfig.AWSRegion != "" { + cfg.Region = redisConfig.AWSRegion + } + + opts := Options{ + Region: redisConfig.AWSRegion, + Serverless: redisConfig.AWSServerless, + } + + provider, err := NewTokenProvider(cfg, redisConfig.AWSCacheName, redisConfig.Username, opts) + if err != nil { + return nil, err + } + if _, err := provider.Token(ctx); err != nil { + return nil, err + } + + resolvedRegion := provider.(*tokenProvider).region //nolint:forcetypeassert + + loggers.Infof("ElastiCache IAM auth enabled (cache=%s, region=%s, serverless=%t)", + redisConfig.AWSCacheName, resolvedRegion, redisConfig.AWSServerless) + + entry := sharedProviderEntry{provider: provider, resolvedRegion: resolvedRegion} + sharedProviders[key] = entry + return provider, nil +} + +func makeSharedTestRedisConfig(cacheName, username, region string, serverless bool) config.RedisConfig { + rc := config.RedisConfig{} + rc.AWSAuth = true + rc.AWSCacheName = cacheName + rc.Username = username + rc.AWSRegion = region + rc.AWSServerless = serverless + return rc +} + +// TestSharedTokenProvider_SameConfigReturnsSameInstance verifies that two calls with +// the same fingerprint return the identical provider instance. +func TestSharedTokenProvider_SameConfigReturnsSameInstance(t *testing.T) { + ResetSharedTokenProvidersForTest() + defer ResetSharedTokenProvidersForTest() + + rc := makeSharedTestRedisConfig("my-cache", "iam-user", "us-east-1", false) + loggers := ldlog.NewDisabledLoggers() + + p1, err := sharedTokenProviderWithConfig(context.Background(), staticCreds(), rc, loggers) + require.NoError(t, err) + require.NotNil(t, p1) + + p2, err := sharedTokenProviderWithConfig(context.Background(), staticCreds(), rc, loggers) + require.NoError(t, err) + + assert.Same(t, p1.(*tokenProvider), p2.(*tokenProvider), + "same fingerprint must return the identical pointer") +} + +// TestSharedTokenProvider_DifferentConfigReturnsDifferentInstance verifies that two +// calls with different cache names return different provider instances. +func TestSharedTokenProvider_DifferentConfigReturnsDifferentInstance(t *testing.T) { + ResetSharedTokenProvidersForTest() + defer ResetSharedTokenProvidersForTest() + + rc1 := makeSharedTestRedisConfig("cache-a", "iam-user", "us-east-1", false) + rc2 := makeSharedTestRedisConfig("cache-b", "iam-user", "us-east-1", false) + loggers := ldlog.NewDisabledLoggers() + + p1, err := sharedTokenProviderWithConfig(context.Background(), staticCreds(), rc1, loggers) + require.NoError(t, err) + + p2, err := sharedTokenProviderWithConfig(context.Background(), staticCreds(), rc2, loggers) + require.NoError(t, err) + + assert.NotSame(t, p1.(*tokenProvider), p2.(*tokenProvider), + "different cache names must return different provider instances") +} + +// TestSharedTokenProvider_ServerlessFingerprintDiffers verifies that serverless=true and +// serverless=false for the same cache produce distinct provider instances. +func TestSharedTokenProvider_ServerlessFingerprintDiffers(t *testing.T) { + ResetSharedTokenProvidersForTest() + defer ResetSharedTokenProvidersForTest() + + rcStd := makeSharedTestRedisConfig("my-cache", "iam-user", "us-east-1", false) + rcSvl := makeSharedTestRedisConfig("my-cache", "iam-user", "us-east-1", true) + loggers := ldlog.NewDisabledLoggers() + + pStd, err := sharedTokenProviderWithConfig(context.Background(), staticCreds(), rcStd, loggers) + require.NoError(t, err) + + pSvl, err := sharedTokenProviderWithConfig(context.Background(), staticCreds(), rcSvl, loggers) + require.NoError(t, err) + + assert.NotSame(t, pStd.(*tokenProvider), pSvl.(*tokenProvider), + "serverless and non-serverless variants must be distinct provider instances") +} + +// TestSharedTokenProvider_ResetHookClearsCache verifies that ResetSharedTokenProvidersForTest +// causes the next call to construct a fresh provider. +func TestSharedTokenProvider_ResetHookClearsCache(t *testing.T) { + ResetSharedTokenProvidersForTest() + + rc := makeSharedTestRedisConfig("my-cache", "iam-user", "us-east-1", false) + loggers := ldlog.NewDisabledLoggers() + + p1, err := sharedTokenProviderWithConfig(context.Background(), staticCreds(), rc, loggers) + require.NoError(t, err) + + ResetSharedTokenProvidersForTest() + + p2, err := sharedTokenProviderWithConfig(context.Background(), staticCreds(), rc, loggers) + require.NoError(t, err) + + assert.NotSame(t, p1.(*tokenProvider), p2.(*tokenProvider), + "after reset, a new provider instance must be created") + + ResetSharedTokenProvidersForTest() +} + +// TestSharedTokenProvider_ProbedOnce verifies that a second call with the same +// fingerprint does not trigger another Token() probe. We test this indirectly: +// a failing credentials provider is replaced with a working one after the first +// call; the second call must use the cached instance (no re-probe that would fail). +func TestSharedTokenProvider_ProbedOnce(t *testing.T) { + ResetSharedTokenProvidersForTest() + defer ResetSharedTokenProvidersForTest() + + rc := makeSharedTestRedisConfig("probe-cache", "iam-user", "us-east-1", false) + loggers := ldlog.NewDisabledLoggers() + + // First call: succeeds. + p1, err := sharedTokenProviderWithConfig(context.Background(), staticCreds(), rc, loggers) + require.NoError(t, err) + + // Second call with a broken cfg — the cache must return p1 without calling Token(). + brokenCfg := aws.Config{ + Region: "us-east-1", + Credentials: &errCredsProvider{err: assert.AnError}, + } + p2, err := sharedTokenProviderWithConfig(context.Background(), brokenCfg, rc, loggers) + require.NoError(t, err, "second call must return cached provider, not re-probe") + assert.Same(t, p1.(*tokenProvider), p2.(*tokenProvider)) +} diff --git a/internal/awsredisauth/token_provider.go b/internal/awsredisauth/token_provider.go new file mode 100644 index 00000000..e8f84913 --- /dev/null +++ b/internal/awsredisauth/token_provider.go @@ -0,0 +1,174 @@ +// Package awsredisauth generates SigV4-presigned authentication tokens for +// AWS ElastiCache IAM authentication. Each Token() call produces a fresh +// presigned URL (with the scheme stripped) that can be passed as the Redis +// AUTH password for a new connection. +// +// Token generation is stateless: no caching, no background refresh goroutines. +// The AWS SDK's credential cache (aws.CredentialsCache) handles Layer 1 (IAM +// credential refresh); this package owns only Layer 2 (per-connection token +// generation via SigV4 presigning). +package awsredisauth + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4" +) + +const ( + // elasticacheSigningService is the SigV4 service identifier for ElastiCache IAM auth. + elasticacheSigningService = "elasticache" + + // emptyPayloadHash is the SHA-256 hash of an empty body, required by PresignHTTP. + emptyPayloadHash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + + // defaultTokenLifetime is the AWS-documented token validity window. + defaultTokenLifetime = 15 * time.Minute +) + +// TokenProvider generates ElastiCache IAM authentication tokens. +type TokenProvider interface { + // Token generates a fresh SigV4-presigned authentication token. + // The returned string is the presigned URL with the scheme stripped + // (e.g. "my-cache/?Action=connect&User=iam-user&X-Amz-..."). + // Callers pass this as the password to Redis AUTH or HELLO. + Token(ctx context.Context) (string, error) +} + +// Options holds optional configuration for NewTokenProvider. +type Options struct { + // Region overrides cfg.Region for SigV4 signing. If empty, cfg.Region is used. + Region string + + // TokenLifetime overrides the default 15-minute token expiry (X-Amz-Expires). + // Primarily useful in tests to shorten the validity window. If zero, defaults + // to 15 minutes (900 seconds). + TokenLifetime time.Duration + + // Serverless indicates that the target cache is an ElastiCache Serverless cache. + // When true, the signed token URL includes the query parameter + // ResourceType=ServerlessCache, which ElastiCache Serverless requires for IAM + // authentication. Without it, the auth request fails with WRONGPASS. + Serverless bool +} + +// tokenProvider is the concrete, stateless implementation of TokenProvider. +type tokenProvider struct { + creds aws.CredentialsProvider + signer *v4.Signer + endpoint string // "https:///" + baseQuery url.Values + region string + expires time.Duration +} + +// NewTokenProvider constructs a stateless TokenProvider. On each Token() call +// it retrieves credentials from cfg.Credentials and presigns a SigV4 URL of +// the form: +// +// https:///?Action=connect&User= +// +// The token returned by Token() is the signed URL with the "https://" scheme +// stripped, as required by ElastiCache IAM auth. +// +// Returns an error immediately if the resolved region is empty — that is a +// startup misconfiguration and is not recoverable without operator intervention. +// +// cacheName is lowercased defensively; callers should already supply lowercase +// per AWS requirements (cache names are converted to lowercase at creation time). +func NewTokenProvider(cfg aws.Config, cacheName, user string, opts ...Options) (TokenProvider, error) { + if cfg.Credentials == nil { + return nil, errors.New("awsredisauth: cfg.Credentials must not be nil") + } + + opt := Options{} + if len(opts) > 0 { + opt = opts[0] + } + + region := opt.Region + if region == "" { + region = cfg.Region + } + if region == "" { + return nil, errors.New("awsredisauth: region is required; set cfg.Region or Options.Region") + } + + lifetime := opt.TokenLifetime + if lifetime == 0 { + lifetime = defaultTokenLifetime + } + + cacheName = strings.ToLower(cacheName) + + // Pre-build the stable base query parameters (Action and User). X-Amz-Expires + // and the SigV4 parameters are added per-call in Token(). + baseQuery := url.Values{} + baseQuery.Set("Action", "connect") + baseQuery.Set("User", user) + if opt.Serverless { + // ElastiCache Serverless requires ResourceType=ServerlessCache in the signed + // token URL; without it the cluster rejects the connection with WRONGPASS. + baseQuery.Set("ResourceType", "ServerlessCache") + } + + return &tokenProvider{ + creds: cfg.Credentials, + signer: v4.NewSigner(), + endpoint: fmt.Sprintf("https://%s/", cacheName), + baseQuery: baseQuery, + region: region, + expires: lifetime, + }, nil +} + +// Token generates a fresh presigned authentication token. It is safe to call +// concurrently because it builds a new http.Request each time and does not +// mutate any shared state. +func (p *tokenProvider) Token(ctx context.Context) (string, error) { + creds, err := p.creds.Retrieve(ctx) + if err != nil { + return "", fmt.Errorf("awsredisauth: retrieving credentials: %w", err) + } + + req, err := http.NewRequest(http.MethodGet, p.endpoint, nil) + if err != nil { + // Should never happen for a valid endpoint built in NewTokenProvider. + return "", fmt.Errorf("awsredisauth: building request: %w", err) + } + + // Clone the base query and add X-Amz-Expires before signing. + q := url.Values{} + for k, v := range p.baseQuery { + q[k] = v + } + q.Set("X-Amz-Expires", strconv.Itoa(int(p.expires.Seconds()))) + req.URL.RawQuery = q.Encode() + + signedURI, _, err := p.signer.PresignHTTP( + ctx, + creds, + req, + emptyPayloadHash, + elasticacheSigningService, + p.region, + time.Now().UTC(), + ) + if err != nil { + return "", fmt.Errorf("awsredisauth: presigning request: %w", err) + } + + // Strip the scheme. ElastiCache expects the token to be the presigned URL + // without the "https://" prefix. + token := strings.TrimPrefix(signedURI, "https://") + token = strings.TrimPrefix(token, "http://") + return token, nil +} diff --git a/internal/awsredisauth/token_provider_test.go b/internal/awsredisauth/token_provider_test.go new file mode 100644 index 00000000..fc79f070 --- /dev/null +++ b/internal/awsredisauth/token_provider_test.go @@ -0,0 +1,297 @@ +package awsredisauth + +import ( + "context" + "errors" + "net/url" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + testRegion = "us-east-1" + testCacheName = "my-cache" + testUser = "iam-user-01" +) + +// staticCreds returns an aws.Config with a static credentials provider +// suitable for deterministic signing in tests. +func staticCreds() aws.Config { + return aws.Config{ + Region: testRegion, + Credentials: credentials.NewStaticCredentialsProvider( + "AKIAIOSFODNN7EXAMPLE", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "", + ), + } +} + +// mustNewProvider is a test helper that calls NewTokenProvider and requires no error. +func mustNewProvider(t *testing.T, cfg aws.Config, cacheName, user string, opts ...Options) TokenProvider { + t.Helper() + p, err := NewTokenProvider(cfg, cacheName, user, opts...) + require.NoError(t, err) + return p +} + +// parseToken parses the token returned by Token() back into a *url.URL. +// The token is a scheme-stripped URL, so we re-add "https://" to parse it. +func parseToken(t *testing.T, token string) *url.URL { + t.Helper() + u, err := url.Parse("https://" + token) + require.NoError(t, err, "token must be a valid URL path with query string") + return u +} + +// TestToken_QueryParams verifies that the generated token contains the expected +// query parameters: Action=connect, User=, X-Amz-Expires=900 (default), +// and a non-empty X-Amz-Signature. The scheme must be https and the host must +// be the cache name. +func TestToken_QueryParams(t *testing.T) { + p := mustNewProvider(t, staticCreds(), testCacheName, testUser) + + token, err := p.Token(context.Background()) + require.NoError(t, err) + require.NotEmpty(t, token, "token must not be empty") + + // Token is a scheme-stripped URL; restore scheme to parse. + u := parseToken(t, token) + + assert.Equal(t, "https", u.Scheme) + assert.Equal(t, testCacheName, u.Hostname(), "host must be the cache name") + + q := u.Query() + assert.Equal(t, "connect", q.Get("Action"), "Action must be 'connect'") + assert.Equal(t, testUser, q.Get("User"), "User must match") + assert.Equal(t, "900", q.Get("X-Amz-Expires"), "default X-Amz-Expires must be 900") + assert.NotEmpty(t, q.Get("X-Amz-Signature"), "X-Amz-Signature must be present") +} + +// TestToken_CustomTokenLifetime verifies that Options.TokenLifetime overrides the +// X-Amz-Expires query parameter in the generated token. +func TestToken_CustomTokenLifetime(t *testing.T) { + p := mustNewProvider(t, staticCreds(), testCacheName, testUser, Options{ + TokenLifetime: 5 * time.Second, + }) + + token, err := p.Token(context.Background()) + require.NoError(t, err) + + u := parseToken(t, token) + assert.Equal(t, "5", u.Query().Get("X-Amz-Expires"), "X-Amz-Expires must reflect custom lifetime") +} + +// TestToken_RegionInCredential verifies that the signing region flows through into +// the X-Amz-Credential scope of the generated token. +// +// X-Amz-Credential takes the form: ///elasticache/aws4_request +func TestToken_RegionInCredential(t *testing.T) { + const customRegion = "eu-west-2" + + cfg := aws.Config{ + Region: customRegion, + Credentials: credentials.NewStaticCredentialsProvider( + "AKIAIOSFODNN7EXAMPLE", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "", + ), + } + + p := mustNewProvider(t, cfg, testCacheName, testUser) + + token, err := p.Token(context.Background()) + require.NoError(t, err) + + u := parseToken(t, token) + credential := u.Query().Get("X-Amz-Credential") + require.NotEmpty(t, credential, "X-Amz-Credential must be present") + + // X-Amz-Credential = AKID/YYYYMMDD//elasticache/aws4_request + assert.True(t, + strings.Contains(credential, "/"+customRegion+"/elasticache/aws4_request"), + "X-Amz-Credential %q must contain //elasticache/aws4_request; got: %s", + credential, credential, + ) +} + +// TestToken_OptionsRegionOverridesCfgRegion verifies that Options.Region takes +// precedence over cfg.Region when both are provided. +func TestToken_OptionsRegionOverridesCfgRegion(t *testing.T) { + const optRegion = "ap-southeast-1" + + cfg := aws.Config{ + Region: "us-east-1", // should be ignored + Credentials: credentials.NewStaticCredentialsProvider( + "AKIAIOSFODNN7EXAMPLE", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "", + ), + } + + p := mustNewProvider(t, cfg, testCacheName, testUser, Options{Region: optRegion}) + + token, err := p.Token(context.Background()) + require.NoError(t, err) + + u := parseToken(t, token) + credential := u.Query().Get("X-Amz-Credential") + assert.True(t, + strings.Contains(credential, "/"+optRegion+"/elasticache/aws4_request"), + "credential must use Options.Region %q; got %s", optRegion, credential, + ) +} + +// TestToken_CredentialsRetrieveErrorPropagates verifies that when the credentials +// provider returns an error, Token() surfaces that error. +func TestToken_CredentialsRetrieveErrorPropagates(t *testing.T) { + sentinelErr := errors.New("no credentials available") + + cfg := aws.Config{ + Region: testRegion, + Credentials: &errCredsProvider{err: sentinelErr}, + } + + p := mustNewProvider(t, cfg, testCacheName, testUser) + + _, err := p.Token(context.Background()) + require.Error(t, err) + assert.ErrorIs(t, err, sentinelErr, "credentials error must be wrapped and propagated") +} + +// TestNewTokenProvider_EmptyRegionErrors verifies that NewTokenProvider returns +// an error at construction time when neither cfg.Region nor Options.Region is set. +func TestNewTokenProvider_EmptyRegionErrors(t *testing.T) { + cfg := aws.Config{ + // Region intentionally empty + Credentials: credentials.NewStaticCredentialsProvider("AKID", "SECRET", ""), + } + + _, err := NewTokenProvider(cfg, testCacheName, testUser) + require.Error(t, err) + assert.Contains(t, err.Error(), "region is required") +} + +// TestNewTokenProvider_NilCredentialsErrors verifies that NewTokenProvider returns +// an error at construction time when cfg.Credentials is nil. +func TestNewTokenProvider_NilCredentialsErrors(t *testing.T) { + cfg := aws.Config{ + Region: testRegion, + Credentials: nil, + } + + _, err := NewTokenProvider(cfg, testCacheName, testUser) + require.Error(t, err) + assert.Contains(t, err.Error(), "Credentials must not be nil") +} + +// TestToken_CacheNameLowercased verifies that the cache name is lowercased +// defensively, so the URL host matches what AWS expects. +func TestToken_CacheNameLowercased(t *testing.T) { + p := mustNewProvider(t, staticCreds(), "My-Cache", testUser) + + token, err := p.Token(context.Background()) + require.NoError(t, err) + + u := parseToken(t, token) + assert.Equal(t, "my-cache", u.Hostname(), "cache name must be lowercased") +} + +// TestToken_PerCallFreshness verifies that calling Token() twice produces two +// non-identical tokens (different X-Amz-Date when called more than 1 second apart, +// or at least that both calls succeed independently). +// +// Note: because the AWS SDK v4 signer does not expose a clock injection seam, +// we cannot deterministically control the signing time. We therefore verify only +// that two sequential calls both succeed and produce non-empty tokens. The +// stateless design guarantees freshness architecturally (no cache means each +// call re-signs). +func TestToken_PerCallFreshness(t *testing.T) { + p := mustNewProvider(t, staticCreds(), testCacheName, testUser) + + tok1, err1 := p.Token(context.Background()) + require.NoError(t, err1) + require.NotEmpty(t, tok1) + + tok2, err2 := p.Token(context.Background()) + require.NoError(t, err2) + require.NotEmpty(t, tok2) + + // Both tokens must be valid signed URLs. + u1 := parseToken(t, tok1) + u2 := parseToken(t, tok2) + assert.NotEmpty(t, u1.Query().Get("X-Amz-Signature")) + assert.NotEmpty(t, u2.Query().Get("X-Amz-Signature")) + + // Note: tok1 == tok2 is possible if both calls happen within the same second + // (same X-Amz-Date). That is acceptable and does not indicate a bug. +} + +// TestToken_Serverless_ResourceTypePresent verifies that when Options.Serverless=true, +// the token URL contains ResourceType=ServerlessCache. +func TestToken_Serverless_ResourceTypePresent(t *testing.T) { + p := mustNewProvider(t, staticCreds(), testCacheName, testUser, Options{Serverless: true}) + + token, err := p.Token(context.Background()) + require.NoError(t, err) + + u := parseToken(t, token) + assert.Equal(t, "ServerlessCache", u.Query().Get("ResourceType"), + "Serverless token must contain ResourceType=ServerlessCache") +} + +// TestToken_Serverless_ResourceTypeAbsent verifies that when Options.Serverless=false +// (the default), the token URL does NOT contain a ResourceType parameter. +func TestToken_Serverless_ResourceTypeAbsent(t *testing.T) { + p := mustNewProvider(t, staticCreds(), testCacheName, testUser) + + token, err := p.Token(context.Background()) + require.NoError(t, err) + + u := parseToken(t, token) + assert.Empty(t, u.Query().Get("ResourceType"), + "non-Serverless token must not contain ResourceType") +} + +// TestToken_OptionsRegionFlowsThroughToCredential verifies that Options.Region appears +// in the X-Amz-Credential scope of the generated token, overriding cfg.Region. +func TestToken_OptionsRegionFlowsThroughToCredential(t *testing.T) { + const overrideRegion = "eu-central-1" + + cfg := aws.Config{ + Region: "us-east-1", // should be overridden + Credentials: credentials.NewStaticCredentialsProvider( + "AKIAIOSFODNN7EXAMPLE", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "", + ), + } + + p := mustNewProvider(t, cfg, testCacheName, testUser, Options{Region: overrideRegion}) + + token, err := p.Token(context.Background()) + require.NoError(t, err) + + u := parseToken(t, token) + credential := u.Query().Get("X-Amz-Credential") + require.NotEmpty(t, credential) + assert.True(t, + strings.Contains(credential, "/"+overrideRegion+"/elasticache/aws4_request"), + "X-Amz-Credential %q must use Options.Region %q", credential, overrideRegion, + ) +} + +// errCredsProvider is a test-only aws.CredentialsProvider that always returns an error. +type errCredsProvider struct { + err error +} + +func (e *errCredsProvider) Retrieve(_ context.Context) (aws.Credentials, error) { + return aws.Credentials{}, e.err +} diff --git a/internal/bigsegments/store_redis.go b/internal/bigsegments/store_redis.go index 876a896a..a4801a25 100644 --- a/internal/bigsegments/store_redis.go +++ b/internal/bigsegments/store_redis.go @@ -7,6 +7,7 @@ import ( "strconv" "github.com/launchdarkly/ld-relay/v8/config" + "github.com/launchdarkly/ld-relay/v8/internal/awsredisauth" "github.com/launchdarkly/ld-relay/v8/internal/sdks" "github.com/launchdarkly/go-sdk-common/v4/ldlog" @@ -79,6 +80,27 @@ func newRedisBigSegmentStore( } } + if redisConfig.AWSAuth { + // Clear any password/username that came in via URL parsing. Otherwise + // go-redis runs a pipeline-AUTH before OnConnect fires and sends the + // static credentials to ElastiCache, which rejects them. + opts.Username = "" + opts.Password = "" + + provider, err := awsredisauth.SharedTokenProvider(context.Background(), redisConfig, loggers) + if err != nil { + return nil, err + } + opts.OnConnect = func(ctx context.Context, cn *redis.Conn) error { + tok, err := provider.Token(ctx) + if err != nil { + return err + } + return cn.AuthACL(ctx, redisConfig.Username, tok).Err() + } + opts.MaxConnAge = awsredisauth.JitteredMaxConnAge() + } + store := redisBigSegmentStore{ client: redis.NewUniversalClient(&opts), prefix: prefix, diff --git a/internal/bigsegments/store_redis_aws_test.go b/internal/bigsegments/store_redis_aws_test.go new file mode 100644 index 00000000..8ad6309a --- /dev/null +++ b/internal/bigsegments/store_redis_aws_test.go @@ -0,0 +1,85 @@ +package bigsegments + +import ( + "context" + "errors" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/launchdarkly/go-configtypes" + "github.com/launchdarkly/go-sdk-common/v4/ldlog" + "github.com/launchdarkly/ld-relay/v8/config" + "github.com/launchdarkly/ld-relay/v8/internal/awsredisauth" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// errCredsProvider is a test-only aws.CredentialsProvider that always returns an error. +type errCredsProvider struct { + err error +} + +func (e *errCredsProvider) Retrieve(_ context.Context) (aws.Credentials, error) { + return aws.Credentials{}, e.err +} + +// staticAWSConfig returns an aws.Config with deterministic static credentials for tests. +func staticAWSConfig() aws.Config { + return aws.Config{ + Region: "us-east-1", + Credentials: credentials.NewStaticCredentialsProvider( + "AKIAIOSFODNN7EXAMPLE", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "", + ), + } +} + +// awsRedisConfig returns a RedisConfig with AWSAuth=true and all required companion fields. +func awsRedisConfig() config.RedisConfig { + rc := config.RedisConfig{} + rc.URL, _ = configtypes.NewOptURLAbsoluteFromString("rediss://my-cache.abc123.use1.cache.amazonaws.com:6379") + rc.AWSAuth = true + rc.AWSCacheName = "my-cache" + rc.Username = "iam-user-01" + rc.TLS = true + return rc +} + +// TestNewRedisBigSegmentStore_AWSAuthCredentialsError verifies that when AWSAuth=true +// and the AWS credentials provider returns an error, newRedisBigSegmentStore returns that +// error at construction time via fail-fast token verification. +func TestNewRedisBigSegmentStore_AWSAuthCredentialsError(t *testing.T) { + sentinelErr := errors.New("no credentials available") + + cfg := aws.Config{ + Region: "us-east-1", + Credentials: &errCredsProvider{err: sentinelErr}, + } + rc := awsRedisConfig() + + _, err := awsredisauth.NewTokenProviderFromAWSConfig(context.Background(), cfg, rc) + require.Error(t, err) + assert.ErrorIs(t, err, sentinelErr) +} + +// TestNewRedisBigSegmentStore_AWSAuthSuccess verifies that when AWSAuth=true and a valid +// aws.Config is available, provider construction succeeds. +func TestNewRedisBigSegmentStore_AWSAuthSuccess(t *testing.T) { + provider, err := awsredisauth.NewTokenProviderFromAWSConfig(context.Background(), staticAWSConfig(), awsRedisConfig()) + require.NoError(t, err) + require.NotNil(t, provider) +} + +// TestNewRedisBigSegmentStore_AWSAuth_ErrorPropagated verifies that newRedisBigSegmentStore +// returns an error when AWSAuth=true and no real AWS credentials are available (CI). +func TestNewRedisBigSegmentStore_AWSAuth_ErrorPropagated(t *testing.T) { + awsredisauth.ResetSharedTokenProvidersForTest() + defer awsredisauth.ResetSharedTokenProvidersForTest() + + rc := awsRedisConfig() + _, err := newRedisBigSegmentStore(rc, config.EnvConfig{}, false, ldlog.NewDisabledLoggers()) + // In CI (no AWS credentials or region), startup must fail fast. + assert.Error(t, err) +} diff --git a/internal/sdks/big_segments.go b/internal/sdks/big_segments.go index 9e6c1fac..8bfa30e9 100644 --- a/internal/sdks/big_segments.go +++ b/internal/sdks/big_segments.go @@ -24,7 +24,10 @@ func ConfigureBigSegments( var storeFactory subsystems.ComponentConfigurer[subsystems.BigSegmentStore] if allConfig.Redis.URL.IsDefined() { - redisBuilder, redisURL := makeRedisDataStoreBuilder(ldredis.BigSegmentStore, allConfig, envConfig) + redisBuilder, redisURL, err := makeRedisDataStoreBuilder(ldredis.BigSegmentStore, allConfig, envConfig, loggers) + if err != nil { + return nil, err + } redactedURL := util.RedactURL(redisURL) loggers.Infof("Using Redis big segment store: %s with prefix: %s", redactedURL, envConfig.Prefix) storeFactory = redisBuilder diff --git a/internal/sdks/data_stores.go b/internal/sdks/data_stores.go index 22d1d27a..3a91897b 100644 --- a/internal/sdks/data_stores.go +++ b/internal/sdks/data_stores.go @@ -6,6 +6,7 @@ import ( "strings" "github.com/launchdarkly/ld-relay/v8/config" + "github.com/launchdarkly/ld-relay/v8/internal/awsredisauth" "github.com/launchdarkly/ld-relay/v8/internal/util" "github.com/launchdarkly/go-sdk-common/v4/ldlog" @@ -57,7 +58,10 @@ func ConfigureDataStore( if allConfig.Redis.URL.IsDefined() { // Our config validation already takes care of normalizing the Redis parameters so that if a // host & port were specified, they are transformed into a URL. - redisBuilder, redisURL := makeRedisDataStoreBuilder(ldredis.DataStore, allConfig, envConfig) + redisBuilder, redisURL, err := makeRedisDataStoreBuilder(ldredis.DataStore, allConfig, envConfig, loggers) + if err != nil { + return nil, DataStoreEnvironmentInfo{}, err + } redactedURL := util.RedactURL(redisURL) loggers.Infof("Using Redis data store: %s with prefix: %s", redactedURL, envConfig.Prefix) @@ -151,22 +155,37 @@ func makeRedisDataStoreBuilder[T any]( constructor func() *ldredis.StoreBuilder[T], allConfig config.Config, envConfig config.EnvConfig, -) (builder *ldredis.StoreBuilder[T], url string) { + loggers ldlog.Loggers, +) (builder *ldredis.StoreBuilder[T], url string, err error) { redisURL, prefix := GetRedisBasicProperties(allConfig.Redis, envConfig) - var dialOptions []redigo.DialOption - if allConfig.Redis.Password != "" { - dialOptions = append(dialOptions, redigo.DialPassword(allConfig.Redis.Password)) - } - if allConfig.Redis.Username != "" { - dialOptions = append(dialOptions, redigo.DialUsername(allConfig.Redis.Username)) - } - b := constructor(). URL(redisURL). - Prefix(prefix). - DialOptions(dialOptions...) - return b, redisURL + Prefix(prefix) + + if allConfig.Redis.AWSAuth { + provider, provErr := awsredisauth.SharedTokenProvider(context.Background(), allConfig.Redis, loggers) + if provErr != nil { + return nil, redisURL, provErr + } + // ElastiCache IAM auth requires the ACL form of AUTH (AUTH ), + // so DialUsername must accompany PasswordProvider — without it, redigo emits + // the single-arg form (AUTH ) and AWS rejects the connection. + b = b.PasswordProvider(provider.Token). + DialOptions(redigo.DialUsername(allConfig.Redis.Username)). + MaxConnLifetime(awsredisauth.JitteredMaxConnAge()) + } else { + var dialOptions []redigo.DialOption + if allConfig.Redis.Password != "" { + dialOptions = append(dialOptions, redigo.DialPassword(allConfig.Redis.Password)) + } + if allConfig.Redis.Username != "" { + dialOptions = append(dialOptions, redigo.DialUsername(allConfig.Redis.Username)) + } + b = b.DialOptions(dialOptions...) + } + + return b, redisURL, nil } // GetDynamoDBBasicProperties transforms the configuration properties to the standard parameters diff --git a/internal/sdks/data_stores_aws_test.go b/internal/sdks/data_stores_aws_test.go new file mode 100644 index 00000000..5e475fa0 --- /dev/null +++ b/internal/sdks/data_stores_aws_test.go @@ -0,0 +1,110 @@ +package sdks + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/launchdarkly/go-configtypes" + "github.com/launchdarkly/go-sdk-common/v4/ldlog" + ldredis "github.com/launchdarkly/go-server-sdk-redis-redigo/v3" + "github.com/launchdarkly/ld-relay/v8/config" + "github.com/launchdarkly/ld-relay/v8/internal/awsredisauth" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// errCredsProvider is a test-only aws.CredentialsProvider that always returns an error. +type errCredsProvider struct { + err error +} + +func (e *errCredsProvider) Retrieve(_ context.Context) (aws.Credentials, error) { + return aws.Credentials{}, e.err +} + +// staticAWSConfig returns an aws.Config with deterministic static credentials for tests. +func staticAWSConfig() aws.Config { + return aws.Config{ + Region: "us-east-1", + Credentials: credentials.NewStaticCredentialsProvider( + "AKIAIOSFODNN7EXAMPLE", + "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "", + ), + } +} + +// awsRedisConfig returns a RedisConfig with AWSAuth=true and required companion fields. +func awsRedisConfig() config.RedisConfig { + rc := config.RedisConfig{} + rc.URL, _ = configtypes.NewOptURLAbsoluteFromString("rediss://my-cache.abc123.use1.cache.amazonaws.com:6379") + rc.AWSAuth = true + rc.AWSCacheName = "my-cache" + rc.Username = "iam-user-01" + rc.TLS = true + return rc +} + +// makeRedisDataStoreBuilderWithProvider is a test helper that bypasses LoadDefaultConfig +// by injecting a pre-constructed TokenProvider directly into the builder. This lets unit +// tests verify the builder wiring without needing real AWS credentials. +func makeRedisDataStoreBuilderWithProvider[T any]( + constructor func() *ldredis.StoreBuilder[T], + allConfig config.Config, + envConfig config.EnvConfig, + provider awsredisauth.TokenProvider, +) (*ldredis.StoreBuilder[T], string) { + redisURL, prefix := GetRedisBasicProperties(allConfig.Redis, envConfig) + b := constructor(). + URL(redisURL). + Prefix(prefix). + PasswordProvider(provider.Token). + MaxConnLifetime(11 * time.Hour) + return b, redisURL +} + +// TestMakeRedisDataStoreBuilder_AWSAuthCredentialsError verifies that when AWSAuth=true +// and the AWS credentials provider returns an error, the fail-fast token verification in +// NewTokenProviderFromAWSConfig surfaces that error at construction time. +func TestMakeRedisDataStoreBuilder_AWSAuthCredentialsError(t *testing.T) { + sentinelErr := errors.New("no credentials available") + + cfg := aws.Config{ + Region: "us-east-1", + Credentials: &errCredsProvider{err: sentinelErr}, + } + rc := awsRedisConfig() + + _, err := awsredisauth.NewTokenProviderFromAWSConfig(context.Background(), cfg, rc) + require.Error(t, err) + assert.ErrorIs(t, err, sentinelErr) +} + +// TestMakeRedisDataStoreBuilder_AWSAuthSuccess verifies that when AWSAuth=true and a valid +// aws.Config is available, a non-nil builder is returned with AWS provider wired in. +func TestMakeRedisDataStoreBuilder_AWSAuthSuccess(t *testing.T) { + provider, err := awsredisauth.NewTokenProviderFromAWSConfig(context.Background(), staticAWSConfig(), awsRedisConfig()) + require.NoError(t, err) + require.NotNil(t, provider) + + allConfig := config.Config{Redis: awsRedisConfig()} + b, _ := makeRedisDataStoreBuilderWithProvider(ldredis.DataStore, allConfig, config.EnvConfig{}, provider) + assert.NotNil(t, b) +} + +// TestConfigureDataStore_AWSAuth_Error verifies that ConfigureDataStore with AWSAuth=true +// returns an error when no real AWS credentials are available (CI environment). This confirms +// the error from makeRedisDataStoreBuilder propagates up through ConfigureDataStore. +func TestConfigureDataStore_AWSAuth_Error(t *testing.T) { + awsredisauth.ResetSharedTokenProvidersForTest() + defer awsredisauth.ResetSharedTokenProvidersForTest() + + allConfig := config.Config{Redis: awsRedisConfig()} + _, _, err := ConfigureDataStore(allConfig, config.EnvConfig{}, ldlog.NewDisabledLoggers()) + // In CI (no AWS credentials or region configured), startup must fail fast. + assert.Error(t, err) +}