diff --git a/docs/content/self-hosting/configuration.mdoc b/docs/content/self-hosting/configuration.mdoc index b0d1317ee..cc07220b0 100644 --- a/docs/content/self-hosting/configuration.mdoc +++ b/docs/content/self-hosting/configuration.mdoc @@ -33,8 +33,8 @@ Choose one message queue provider. The selected provider is used for both event | Variable | Description | |----------|-------------| -| `AWS_SQS_ACCESS_KEY_ID` | AWS Access Key ID | -| `AWS_SQS_SECRET_ACCESS_KEY` | AWS Secret Access Key | +| `AWS_SQS_ACCESS_KEY_ID` | AWS Access Key ID (optional; omit to use the AWS SDK default credential chain, e.g. an IAM role) | +| `AWS_SQS_SECRET_ACCESS_KEY` | AWS Secret Access Key (optional; omit to use the AWS SDK default credential chain, e.g. an IAM role) | | `AWS_SQS_REGION` | AWS Region | **GCP Pub/Sub:** diff --git a/internal/config/mqconfig_aws.go b/internal/config/mqconfig_aws.go index 69bd6f0e0..b3ae1a89d 100644 --- a/internal/config/mqconfig_aws.go +++ b/internal/config/mqconfig_aws.go @@ -2,6 +2,7 @@ package config import ( "context" + "errors" "fmt" "time" @@ -9,9 +10,12 @@ import ( "github.com/hookdeck/outpost/internal/mqs" ) +// errPartialAWSSQSCredentials rejects a config with exactly one static key set. +var errPartialAWSSQSCredentials = errors.New("AWS SQS: both access_key_id and secret_access_key must be set together, or both omitted to use the default credential chain") + type AWSSQSConfig struct { - AccessKeyID string `yaml:"access_key_id" env:"AWS_SQS_ACCESS_KEY_ID" desc:"AWS Access Key ID for SQS. Required if AWS SQS is the chosen MQ provider." required:"C"` - SecretAccessKey string `yaml:"secret_access_key" env:"AWS_SQS_SECRET_ACCESS_KEY" desc:"AWS Secret Access Key for SQS. Required if AWS SQS is the chosen MQ provider." required:"C"` + AccessKeyID string `yaml:"access_key_id" env:"AWS_SQS_ACCESS_KEY_ID" desc:"AWS Access Key ID for SQS. Optional: omit (with the secret access key) to use the AWS SDK default credential chain, e.g. an IAM role." required:"N"` + SecretAccessKey string `yaml:"secret_access_key" env:"AWS_SQS_SECRET_ACCESS_KEY" desc:"AWS Secret Access Key for SQS. Optional: omit (with the access key ID) to use the AWS SDK default credential chain, e.g. an IAM role." required:"N"` Region string `yaml:"region" env:"AWS_SQS_REGION" desc:"AWS Region for SQS. Required if AWS SQS is the chosen MQ provider." required:"C"` Endpoint string `yaml:"endpoint" env:"AWS_SQS_ENDPOINT" desc:"Custom AWS SQS endpoint URL. Optional, typically used for local testing (e.g., LocalStack)." required:"N"` DeliveryQueue string `yaml:"delivery_queue" env:"AWS_SQS_DELIVERY_QUEUE" desc:"Name of the SQS queue for delivery events." required:"N"` @@ -45,6 +49,9 @@ func (c *AWSSQSConfig) ToInfraConfig(queueType string) *mqinfra.MQInfraConfig { } func (c *AWSSQSConfig) ToQueueConfig(ctx context.Context, queueType string) (*mqs.QueueConfig, error) { + if (c.AccessKeyID == "") != (c.SecretAccessKey == "") { + return nil, errPartialAWSSQSCredentials + } return &mqs.QueueConfig{ AWSSQS: &mqs.AWSSQSConfig{ Endpoint: c.Endpoint, @@ -60,6 +67,8 @@ func (c *AWSSQSConfig) GetProviderType() string { return "awssqs" } +// IsConfigured selects SQS on Region alone; keys are optional so IAM-role auth +// can use the AWS SDK default credential chain. func (c *AWSSQSConfig) IsConfigured() bool { - return c.AccessKeyID != "" && c.SecretAccessKey != "" && c.Region != "" + return c.Region != "" } diff --git a/internal/config/mqconfig_aws_test.go b/internal/config/mqconfig_aws_test.go new file mode 100644 index 000000000..cbf2c2d6b --- /dev/null +++ b/internal/config/mqconfig_aws_test.go @@ -0,0 +1,90 @@ +package config_test + +import ( + "testing" + + "github.com/hookdeck/outpost/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestAWSSQSConfig_IsConfigured(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + cfg config.AWSSQSConfig + want bool + }{ + { + name: "region only (IAM role via default credential chain)", + cfg: config.AWSSQSConfig{Region: "us-east-1"}, + want: true, + }, + { + name: "region with static keys", + cfg: config.AWSSQSConfig{AccessKeyID: "AKID", SecretAccessKey: "SECRET", Region: "us-east-1"}, + want: true, + }, + { + name: "keys without region", + cfg: config.AWSSQSConfig{AccessKeyID: "AKID", SecretAccessKey: "SECRET"}, + want: false, + }, + { + name: "empty", + cfg: config.AWSSQSConfig{}, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tt.want, tt.cfg.IsConfigured()) + }) + } +} + +func TestAWSSQSConfig_ToQueueConfig(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + cfg config.AWSSQSConfig + wantErr bool + }{ + { + name: "both keys set (static credentials)", + cfg: config.AWSSQSConfig{AccessKeyID: "AKID", SecretAccessKey: "SECRET", Region: "us-east-1"}, + }, + { + name: "both keys empty (default credential chain)", + cfg: config.AWSSQSConfig{Region: "us-east-1"}, + }, + { + name: "only access key set", + cfg: config.AWSSQSConfig{AccessKeyID: "AKID", Region: "us-east-1"}, + wantErr: true, + }, + { + name: "only secret key set", + cfg: config.AWSSQSConfig{SecretAccessKey: "SECRET", Region: "us-east-1"}, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + cfg, err := tt.cfg.ToQueueConfig(t.Context(), "deliverymq") + if tt.wantErr { + require.Error(t, err) + assert.Nil(t, cfg) + return + } + require.NoError(t, err) + require.NotNil(t, cfg) + }) + } +} diff --git a/internal/config/publishmq.go b/internal/config/publishmq.go index d77d4172b..68d3261fc 100644 --- a/internal/config/publishmq.go +++ b/internal/config/publishmq.go @@ -7,8 +7,8 @@ import ( ) type PublishAWSSQSConfig struct { - AccessKeyID string `yaml:"access_key_id" env:"PUBLISH_AWS_SQS_ACCESS_KEY_ID" desc:"AWS Access Key ID for the SQS publish queue. Required if AWS SQS is the chosen publish MQ provider." required:"C"` - SecretAccessKey string `yaml:"secret_access_key" env:"PUBLISH_AWS_SQS_SECRET_ACCESS_KEY" desc:"AWS Secret Access Key for the SQS publish queue. Required if AWS SQS is the chosen publish MQ provider." required:"C"` + AccessKeyID string `yaml:"access_key_id" env:"PUBLISH_AWS_SQS_ACCESS_KEY_ID" desc:"AWS Access Key ID for the SQS publish queue. Optional: omit (with the secret access key) to use the AWS SDK default credential chain, e.g. an IAM role." required:"N"` + SecretAccessKey string `yaml:"secret_access_key" env:"PUBLISH_AWS_SQS_SECRET_ACCESS_KEY" desc:"AWS Secret Access Key for the SQS publish queue. Optional: omit (with the access key ID) to use the AWS SDK default credential chain, e.g. an IAM role." required:"N"` Region string `yaml:"region" env:"PUBLISH_AWS_SQS_REGION" desc:"AWS Region for the SQS publish queue. Required if AWS SQS is the chosen publish MQ provider." required:"C"` Endpoint string `yaml:"endpoint" env:"PUBLISH_AWS_SQS_ENDPOINT" desc:"Custom AWS SQS endpoint URL for the publish queue. Optional." required:"N"` Queue string `yaml:"queue" env:"PUBLISH_AWS_SQS_QUEUE" desc:"Name of the SQS queue for publishing events. Required if AWS SQS is the chosen publish MQ provider." required:"C"` @@ -99,9 +99,20 @@ func (c *PublishMQConfig) GetQueueConfig() *mqs.QueueConfig { } } +// Validate enforces the AWS SQS partial-credential rule for the selected provider. +func (c *PublishMQConfig) Validate() error { + if c.GetInfraType() == "awssqs" { + if (c.AWSSQS.AccessKeyID == "") != (c.AWSSQS.SecretAccessKey == "") { + return errPartialAWSSQSCredentials + } + } + return nil +} + +// hasPublishAWSSQSConfig selects SQS on Region alone; keys are optional so +// IAM-role auth can use the AWS SDK default credential chain. func hasPublishAWSSQSConfig(config PublishAWSSQSConfig) bool { - return config.AccessKeyID != "" && - config.SecretAccessKey != "" && config.Region != "" + return config.Region != "" } func hasPublishAzureServiceBusConfig(config PublishAzureServiceBusConfig) bool { diff --git a/internal/config/publishmq_test.go b/internal/config/publishmq_test.go new file mode 100644 index 000000000..700e38419 --- /dev/null +++ b/internal/config/publishmq_test.go @@ -0,0 +1,93 @@ +package config_test + +import ( + "testing" + + "github.com/hookdeck/outpost/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestPublishMQConfig_GetInfraType_AWSSQS(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + cfg config.PublishAWSSQSConfig + want string + }{ + { + name: "region only (IAM role via default credential chain)", + cfg: config.PublishAWSSQSConfig{Region: "us-east-1"}, + want: "awssqs", + }, + { + name: "region with static keys", + cfg: config.PublishAWSSQSConfig{AccessKeyID: "AKID", SecretAccessKey: "SECRET", Region: "us-east-1"}, + want: "awssqs", + }, + { + name: "keys without region", + cfg: config.PublishAWSSQSConfig{AccessKeyID: "AKID", SecretAccessKey: "SECRET"}, + want: "", + }, + { + name: "empty", + cfg: config.PublishAWSSQSConfig{}, + want: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + cfg := config.PublishMQConfig{AWSSQS: tt.cfg} + assert.Equal(t, tt.want, cfg.GetInfraType()) + }) + } +} + +func TestPublishMQConfig_Validate(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + cfg config.PublishMQConfig + wantErr bool + }{ + { + name: "aws sqs both keys set", + cfg: config.PublishMQConfig{AWSSQS: config.PublishAWSSQSConfig{AccessKeyID: "AKID", SecretAccessKey: "SECRET", Region: "us-east-1"}}, + }, + { + name: "aws sqs both keys empty (default credential chain)", + cfg: config.PublishMQConfig{AWSSQS: config.PublishAWSSQSConfig{Region: "us-east-1"}}, + }, + { + name: "aws sqs only access key set", + cfg: config.PublishMQConfig{AWSSQS: config.PublishAWSSQSConfig{AccessKeyID: "AKID", Region: "us-east-1"}}, + wantErr: true, + }, + { + name: "aws sqs only secret key set", + cfg: config.PublishMQConfig{AWSSQS: config.PublishAWSSQSConfig{SecretAccessKey: "SECRET", Region: "us-east-1"}}, + wantErr: true, + }, + { + name: "no publish provider configured", + cfg: config.PublishMQConfig{}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + err := tt.cfg.Validate() + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + }) + } +} diff --git a/internal/config/validation.go b/internal/config/validation.go index 377c65d80..8347b8720 100644 --- a/internal/config/validation.go +++ b/internal/config/validation.go @@ -29,6 +29,10 @@ func (c *Config) Validate(flags Flags) error { return err } + if err := c.validatePublishMQ(); err != nil { + return err + } + if err := c.validateAESEncryptionSecret(); err != nil { return err } @@ -134,6 +138,14 @@ func (c *Config) validateMQs() error { return nil } +// validatePublishMQ validates the optional publish MQ configuration +func (c *Config) validatePublishMQ() error { + if err := c.PublishMQ.Validate(); err != nil { + return fmt.Errorf("failed to validate publish queue config: %w", err) + } + return nil +} + // validateAESEncryptionSecret validates the AES encryption secret func (c *Config) validateAESEncryptionSecret() error { if c.AESEncryptionSecret == "" { diff --git a/internal/mqinfra/mqinfra_test.go b/internal/mqinfra/mqinfra_test.go index 3d9eda8b2..dafd94f5a 100644 --- a/internal/mqinfra/mqinfra_test.go +++ b/internal/mqinfra/mqinfra_test.go @@ -275,6 +275,97 @@ func TestIntegrationMQInfra_AWSSQS(t *testing.T) { ) } +// TestIntegrationMQInfra_AWSSQS_DefaultCredentialChain verifies that when no static +// ServiceAccountCredentials are configured, the AWS SDK default credential chain is +// used instead. Here the chain resolves credentials from environment variables (the +// same mechanism an ECS/EKS task role or EC2 instance profile relies on further down +// the chain). LocalStack accepts any credentials, so this proves the empty-credentials +// path builds a working client and can declare/publish/receive. +func TestIntegrationMQInfra_AWSSQS_DefaultCredentialChain(t *testing.T) { + testutil.CheckIntegrationTest(t) + + // Provide credentials via the environment so the SDK default chain resolves them. + // t.Setenv is incompatible with t.Parallel, so this test runs serially. + t.Setenv("AWS_ACCESS_KEY_ID", "test") + t.Setenv("AWS_SECRET_ACCESS_KEY", "test") + + endpoint := testinfra.EnsureLocalStack() + t.Cleanup(testinfra.Start(t)) + + q := idgen.String() + infraCfg := mqinfra.MQInfraConfig{ + AWSSQS: &mqinfra.AWSSQSInfraConfig{ + Endpoint: endpoint, + ServiceAccountCredentials: "", // no static creds → default credential chain + Region: "us-east-1", + Topic: q, + }, + Policy: mqinfra.Policy{ + RetryLimit: retryLimit, + VisibilityTimeout: 1, + }, + } + mqCfg := mqs.QueueConfig{ + AWSSQS: &mqs.AWSSQSConfig{ + Endpoint: endpoint, + ServiceAccountCredentials: "", // no static creds → default credential chain + Region: "us-east-1", + Topic: q, + WaitTime: 1 * time.Second, + }, + } + + ctx := context.Background() + infra := mqinfra.New(&infraCfg) + require.NoError(t, infra.Declare(ctx)) + t.Cleanup(func() { + require.NoError(t, infra.TearDown(ctx)) + }) + + exists, err := infra.Exist(ctx) + require.NoError(t, err) + require.True(t, exists) + + mq := mqs.NewQueue(&mqCfg) + cleanup, err := mq.Init(ctx) + require.NoError(t, err) + t.Cleanup(cleanup) + + subscription, err := mq.Subscribe(ctx) + require.NoError(t, err) + t.Cleanup(func() { + subscription.Shutdown(ctx) + }) + + msgchan := make(chan *testutil.MockMsg) + go func() { + for { + msg, err := subscription.Receive(ctx) + if err != nil { + log.Println(err) + return + } + msg.Ack() + mockMsg := &testutil.MockMsg{} + if err := mockMsg.FromMessage(msg); err != nil { + log.Println("Error parsing message", err) + } else { + msgchan <- mockMsg + } + } + }() + + msg := &testutil.MockMsg{ID: idgen.String()} + require.NoError(t, mq.Publish(ctx, msg)) + + select { + case receivedMsg := <-msgchan: + assert.Equal(t, msg.ID, receivedMsg.ID) + case <-time.After(1 * time.Second): + require.Fail(t, "timeout waiting for message") + } +} + func TestIntegrationMQInfra_GCPPubSub(t *testing.T) { testutil.CheckIntegrationTest(t) // Set PUBSUB_EMULATOR_HOST environment variable diff --git a/internal/mqs/queue_awssqs.go b/internal/mqs/queue_awssqs.go index 0e1123023..d1b534db1 100644 --- a/internal/mqs/queue_awssqs.go +++ b/internal/mqs/queue_awssqs.go @@ -29,7 +29,14 @@ type AWSSQSConfig struct { WaitTime time.Duration // optional - defaults to 20s if not set } +// ToCredentials returns a static credentials provider, or (nil, nil) when no +// static credentials are set so the caller falls back to the AWS SDK default +// credential chain. func (c *AWSSQSConfig) ToCredentials() (*credentials.StaticCredentialsProvider, error) { + // An empty or all-empty ("::") credential string means no static credentials. + if strings.Trim(c.ServiceAccountCredentials, ":") == "" { + return nil, nil + } creds := strings.Split(c.ServiceAccountCredentials, ":") if len(creds) != 3 { return nil, errors.New("invalid AWS Service Account Credentials") @@ -100,10 +107,14 @@ func (q *AWSQueue) InitSDK(ctx context.Context) error { return err } - sdkConfig, err := config.LoadDefaultConfig(ctx, + opts := []func(*config.LoadOptions) error{ config.WithRegion(q.config.Region), - config.WithCredentialsProvider(creds), - ) + } + if creds != nil { + opts = append(opts, config.WithCredentialsProvider(creds)) + } + + sdkConfig, err := config.LoadDefaultConfig(ctx, opts...) if err != nil { return err } diff --git a/internal/mqs/queue_awssqs_test.go b/internal/mqs/queue_awssqs_test.go new file mode 100644 index 000000000..06cb7c564 --- /dev/null +++ b/internal/mqs/queue_awssqs_test.go @@ -0,0 +1,71 @@ +package mqs_test + +import ( + "testing" + + "github.com/hookdeck/outpost/internal/mqs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestAWSSQSConfig_ToCredentials(t *testing.T) { + t.Parallel() + + t.Run("valid static credentials", func(t *testing.T) { + t.Parallel() + cfg := &mqs.AWSSQSConfig{ServiceAccountCredentials: "AKID:SECRET:"} + creds, err := cfg.ToCredentials() + require.NoError(t, err) + require.NotNil(t, creds) + + value, err := creds.Retrieve(t.Context()) + require.NoError(t, err) + assert.Equal(t, "AKID", value.AccessKeyID) + assert.Equal(t, "SECRET", value.SecretAccessKey) + assert.Empty(t, value.SessionToken) + }) + + t.Run("valid static credentials with session token", func(t *testing.T) { + t.Parallel() + cfg := &mqs.AWSSQSConfig{ServiceAccountCredentials: "AKID:SECRET:TOKEN"} + creds, err := cfg.ToCredentials() + require.NoError(t, err) + require.NotNil(t, creds) + + value, err := creds.Retrieve(t.Context()) + require.NoError(t, err) + assert.Equal(t, "TOKEN", value.SessionToken) + }) + + t.Run("empty string defers to default credential chain", func(t *testing.T) { + t.Parallel() + cfg := &mqs.AWSSQSConfig{ServiceAccountCredentials: ""} + creds, err := cfg.ToCredentials() + require.NoError(t, err) + assert.Nil(t, creds, "empty credentials should return nil so the SDK default chain is used") + }) + + t.Run("all-empty parts defers to default credential chain", func(t *testing.T) { + t.Parallel() + cfg := &mqs.AWSSQSConfig{ServiceAccountCredentials: "::"} + creds, err := cfg.ToCredentials() + require.NoError(t, err) + assert.Nil(t, creds, "\"::\" should be treated as no credentials") + }) + + t.Run("partial credentials still build a static provider", func(t *testing.T) { + t.Parallel() + cfg := &mqs.AWSSQSConfig{ServiceAccountCredentials: "AKID::"} + creds, err := cfg.ToCredentials() + require.NoError(t, err) + require.NotNil(t, creds) + }) + + t.Run("malformed non-empty credentials error", func(t *testing.T) { + t.Parallel() + cfg := &mqs.AWSSQSConfig{ServiceAccountCredentials: "AKID:SECRET"} + creds, err := cfg.ToCredentials() + require.Error(t, err) + assert.Nil(t, creds) + }) +} diff --git a/internal/util/awsutil/awsutil.go b/internal/util/awsutil/awsutil.go index 3803d34f1..c697e5b02 100644 --- a/internal/util/awsutil/awsutil.go +++ b/internal/util/awsutil/awsutil.go @@ -18,10 +18,14 @@ func SQSClientFromConfig(ctx context.Context, cfg *mqs.AWSSQSConfig) (*sqs.Clien return nil, err } - sdkConfig, err := config.LoadDefaultConfig(ctx, + opts := []func(*config.LoadOptions) error{ config.WithRegion(cfg.Region), - config.WithCredentialsProvider(creds), - ) + } + if creds != nil { + opts = append(opts, config.WithCredentialsProvider(creds)) + } + + sdkConfig, err := config.LoadDefaultConfig(ctx, opts...) if err != nil { return nil, err }