feat(mqconfig): support IAM role auth for SQS MQ provider

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:**

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 != ""
 }

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())
+		})
+	}
+}

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

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
 	}

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)
+	})
+}

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
 	}