From e4ee07782ae3ae9fc975afcaf501b0a13550c9a7 Mon Sep 17 00:00:00 2001 From: Ambrozie Paval Date: Fri, 17 Jul 2026 16:53:37 +0300 Subject: [PATCH] feat(mqconfig): support IAM role auth for SQS MQ provider The AWS SQS message-queue provider previously required static IAM user keys (AWS_SQS_ACCESS_KEY_ID / AWS_SQS_SECRET_ACCESS_KEY) and always passed them explicitly to the SDK, which suppressed the default credential chain. This made role-based auth (ECS/EKS task role, EC2 instance profile) impossible. When no static credentials are configured, ToCredentials now returns (nil, nil) and the client builders omit WithCredentialsProvider, letting the AWS SDK default credential chain resolve credentials. IsConfigured selects SQS on region alone so the keys can be omitted. Existing static-key configs are unchanged. Mirrors the GCP Pub/Sub provider's implicit-credential behavior. Refs hookdeck/outpost#463 Generated with AI Co-Authored-By: Claude --- docs/content/self-hosting/configuration.mdoc | 4 +- internal/config/mqconfig_aws.go | 8 +- internal/config/mqconfig_aws_test.go | 46 ++++++++++ internal/mqinfra/mqinfra_test.go | 91 ++++++++++++++++++++ internal/mqs/queue_awssqs.go | 17 +++- internal/mqs/queue_awssqs_test.go | 71 +++++++++++++++ internal/util/awsutil/awsutil.go | 10 ++- 7 files changed, 236 insertions(+), 11 deletions(-) create mode 100644 internal/config/mqconfig_aws_test.go create mode 100644 internal/mqs/queue_awssqs_test.go 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..9b5eebdbf 100644 --- a/internal/config/mqconfig_aws.go +++ b/internal/config/mqconfig_aws.go @@ -10,8 +10,8 @@ import ( ) 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"` @@ -60,6 +60,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..c15fd239e --- /dev/null +++ b/internal/config/mqconfig_aws_test.go @@ -0,0 +1,46 @@ +package config_test + +import ( + "testing" + + "github.com/hookdeck/outpost/internal/config" + "github.com/stretchr/testify/assert" +) + +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()) + }) + } +} 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 }