[TT-15724] - support for aws static credentials

README.md

@@ -1358,7 +1358,13 @@
 
 `region` - The AWS region your Kinesis stream is located - i.e. eu-west-2
 
-`batch_size` - Optional. The maximum size of the records in a batch is 5MiB. If your records are larger in size setting this batch size paramter can guarantee you don't have failed delivery due to too large a batch. Default size if unset is 100.
+`batch_size` - The maximum size of the records in a batch is 5MiB. If your records are larger in size setting this batch size paramter can guarantee you don't have failed delivery due to too large a batch. Default size if unset is 100 (optional)
+
+`access_key_id` - AWS Access Key ID for authentication. If not provided, will use default credential chain (environment variables, shared credentials file, IAM roles, etc.) (optional)
+
+`secret_access_key` - AWS Secret Access Key for authentication. If not provided, will use default credential chain (optional)
+
+`session_token` - AWS Session Token for temporary credentials (optional)
 
 ###### JSON / Conf File
 
@@ -1368,7 +1374,10 @@
       "meta": {
         "stream_name": "my-stream",
         "region": "eu-west-2",
-        "batch_size": 100
+        "batch_size": 100,
+        "access_key_id": "your-key-id",
+        "secret_access_key": "your-secret-key",
+        "session_token": "your-session-token"
       }
     },
 ```
@@ -1376,11 +1385,14 @@
 ###### Env Variables
 
 ```
 #Kinesis Pump Configuration
 TYK_PMP_PUMPS_KINESIS_TYPE=kinesis
-TYK_PMP_PUMPS_KINESIs_META_STREAMNAME=my-stream
+TYK_PMP_PUMPS_KINESIS_META_STREAMNAME=my-stream
 TYK_PMP_PUMPS_KINESIS_META_REGION=eu-west-2
 TYK_PMP_PUMPS_KINESIS_META_BATCHSIZE=100
+TYK_PMP_PUMPS_KINESIS_META_ACCESSKEYID=your-key-id
+TYK_PMP_PUMPS_KINESIS_META_SECRETACCESSKEY=your-secret-key
+TYK_PMP_PUMPS_KINESIS_META_SESSIONTOKEN=your-session-token
 ```
 
 # Base Pump Configurations

pumps/kinesis.go

@@ -13,6 +13,7 @@
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	awsconfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
 	"github.com/aws/aws-sdk-go-v2/service/kinesis"
 	"github.com/aws/aws-sdk-go-v2/service/kinesis/types"
 )
@@ -37,6 +38,12 @@
 	StreamName string `mapstructure:"stream_name"`
 	// AWS Region the Kinesis stream targets
 	Region string `mapstructure:"region"`
+	// AWS Access Key ID for authentication. If not provided, will use default credential chain (environment variables, shared credentials file, IAM roles, etc.) -- Optional
+	AccessKeyID string `mapstructure:"access_key_id"`
+	// AWS Secret Access Key for authentication. If not provided, will use default credential chain -- Optional
+	SecretAccessKey string `mapstructure:"secret_access_key"`
+	// AWS Session Token for temporary credentials -- Optional
+	SessionToken string `mapstructure:"session_token"`
 	// Each PutRecords (the function used in this pump)request can support up to 500 records.
 	// Each record in the request can be as large as 1 MiB, up to a limit of 5 MiB for the entire request, including partition keys.
 	// Each shard can support writes up to 1,000 records per second, up to a maximum data write total of 1 MiB per second.
@@ -68,8 +75,14 @@
 
 	// Load AWS configuration
 	// Credentials are loaded as specified in
 	// https://aws.github.io/aws-sdk-go-v2/docs/configuring-sdk/#specifying-credentials
-	cfg, err := awsconfig.LoadDefaultConfig(context.TODO(), awsconfig.WithRegion(p.kinesisConf.Region))
+	var cfg aws.Config
+	if p.kinesisConf.AccessKeyID != "" && p.kinesisConf.SecretAccessKey != "" {
+		creds := credentials.NewStaticCredentialsProvider(p.kinesisConf.AccessKeyID, p.kinesisConf.SecretAccessKey, p.kinesisConf.SessionToken)
+		cfg, err = awsconfig.LoadDefaultConfig(context.TODO(), awsconfig.WithCredentialsProvider(creds), awsconfig.WithRegion(p.kinesisConf.Region))
+	} else {
+		cfg, err = awsconfig.LoadDefaultConfig(context.TODO(), awsconfig.WithRegion(p.kinesisConf.Region))
+	}
 	if err != nil {
 		p.log.Fatalf("unable to load Kinesis SDK config, %v", err)
 	}

pumps/kinesis_test.go

@@ -0,0 +1,113 @@
+package pumps
+
+import (
+	"testing"
+
+	"github.com/mitchellh/mapstructure"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestKinesisPump_StaticCredentials_ConfigurationParsing(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    map[string]interface{}
+		expected KinesisConf
+	}{
+		{
+			name: "Complete static credentials",
+			input: map[string]interface{}{
+				"stream_name":       "test-stream",
+				"region":            "us-east-1",
+				"access_key_id":     "AKIATEST123",
+				"secret_access_key": "secretkey123",
+				"session_token":     "sessiontoken123",
+				"batch_size":        100,
+			},
+			expected: KinesisConf{
+				StreamName:      "test-stream",
+				Region:          "us-east-1",
+				AccessKeyID:     "AKIATEST123",
+				SecretAccessKey: "secretkey123",
+				SessionToken:    "sessiontoken123",
+				BatchSize:       100,
+			},
+		},
+		{
+			name: "Static credentials without session token",
+			input: map[string]interface{}{
+				"stream_name":       "test-stream",
+				"region":            "us-east-1",
+				"access_key_id":     "AKIATEST123",
+				"secret_access_key": "secretkey123",
+			},
+			expected: KinesisConf{
+				StreamName:      "test-stream",
+				Region:          "us-east-1",
+				AccessKeyID:     "AKIATEST123",
+				SecretAccessKey: "secretkey123",
+				SessionToken:    "",
+			},
+		},
+		{
+			name: "No static credentials (default chain)",
+			input: map[string]interface{}{
+				"stream_name": "test-stream",
+				"region":      "us-west-2",
+			},
+			expected: KinesisConf{
+				StreamName:      "test-stream",
+				Region:          "us-west-2",
+				AccessKeyID:     "",
+				SecretAccessKey: "",
+				SessionToken:    "",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var conf KinesisConf
+			err := mapstructure.Decode(tt.input, &conf)
+			assert.NoError(t, err)
+
+			assert.Equal(t, tt.expected.AccessKeyID, conf.AccessKeyID)
+			assert.Equal(t, tt.expected.SecretAccessKey, conf.SecretAccessKey)
+			assert.Equal(t, tt.expected.SessionToken, conf.SessionToken)
+		})
+	}
+}
+
+func TestKinesisPump_StaticCredentials_Logic(t *testing.T) {
+	t.Run("Should use static credentials when both keys provided", func(t *testing.T) {
+		config := map[string]interface{}{
+			"stream_name":       "test-stream",
+			"region":            "us-east-1",
+			"access_key_id":     "AKIATEST123",
+			"secret_access_key": "secretkey123",
+		}
+
+		pump := &KinesisPump{}
+		pump.Init(config)
+
+		if pump.kinesisConf != nil {
+			hasStaticCreds := pump.kinesisConf.AccessKeyID != "" && pump.kinesisConf.SecretAccessKey != ""
+			assert.True(t, hasStaticCreds, "Should have static credentials")
+		}
+	})
+
+	t.Run("Should fall back to default chain when incomplete", func(t *testing.T) {
+		config := map[string]interface{}{
+			"stream_name":   "test-stream",
+			"region":        "us-east-1",
+			"access_key_id": "AKIATEST123",
+		}
+
+		pump := &KinesisPump{}
+		pump.Init(config)
+
+		if pump.kinesisConf != nil {
+			hasStaticCreds := pump.kinesisConf.AccessKeyID != "" && pump.kinesisConf.SecretAccessKey != ""
+			assert.False(t, hasStaticCreds, "Should use default credential chain")
+		}
+	})
+}