fix(aws): reject incomplete static credential configuration

catalog/glue/glue.go

@@ -32,6 +32,7 @@ import (
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/catalog"
 	"github.com/apache/iceberg-go/catalog/internal"
+	internalaws "github.com/apache/iceberg-go/internal/awsconfig"
 	"github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/metrics"
 	"github.com/apache/iceberg-go/table"
@@ -123,7 +124,10 @@ func toAwsConfig(ctx context.Context, p iceberg.Properties) (aws.Config, error)
 	}
 
 	key, secret, token := p[AccessKeyID], p[SecretAccessKey], p[SessionToken]
-	if key != "" || secret != "" || token != "" {
+	if err := internalaws.ValidateStaticCredentials(AccessKeyID, SecretAccessKey, SessionToken, key, secret, token); err != nil {
+		return aws.Config{}, err
+	}
+	if key != "" {
 		opts = append(opts, config.WithCredentialsProvider(
 			credentials.NewStaticCredentialsProvider(key, secret, token)))
 	}

catalog/glue/glue_test.go

@@ -50,6 +50,40 @@ import (
 
 var errGluePurgeRemove = errors.New("glue purge remove failed")
 
+func TestLoadAWSConfigRejectsIncompleteStaticCredentials(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name  string
+		props iceberg.Properties
+		err   string
+	}{
+		{"access key only", iceberg.Properties{AccessKeyID: "access"}, "glue.access-key-id and glue.secret-access-key must be configured together"},
+		{"secret key only", iceberg.Properties{SecretAccessKey: "secret"}, "glue.access-key-id and glue.secret-access-key must be configured together"},
+		{"session token only", iceberg.Properties{SessionToken: "token"}, "glue.session-token requires glue.access-key-id and glue.secret-access-key"},
+		{"access key and token", iceberg.Properties{AccessKeyID: "access", SessionToken: "token"}, "glue.access-key-id and glue.secret-access-key must be configured together"},
+		{"secret key and token", iceberg.Properties{SecretAccessKey: "secret", SessionToken: "token"}, "glue.access-key-id and glue.secret-access-key must be configured together"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			_, err := toAwsConfig(context.Background(), tt.props)
+			require.ErrorContains(t, err, tt.err)
+		})
+	}
+
+	cfg, err := toAwsConfig(context.Background(), iceberg.Properties{
+		AccessKeyID: "access", SecretAccessKey: "secret", SessionToken: "token",
+	})
+	require.NoError(t, err)
+	creds, err := cfg.Credentials.Retrieve(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, "access", creds.AccessKeyID)
+	require.Equal(t, "secret", creds.SecretAccessKey)
+	require.Equal(t, "token", creds.SessionToken)
+}
+
 type mockGlueClient struct {
 	mock.Mock
 }

internal/awsconfig/credentials.go

@@ -0,0 +1,38 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package awsconfig
+
+import (
+	"errors"
+	"fmt"
+)
+
+var ErrIncompleteStaticCredentials = errors.New("incomplete static AWS credentials")
+
+// ValidateStaticCredentials ensures a session token is only configured with
+// the complete key pair required by AWS static credential providers.
+func ValidateStaticCredentials(keyName, secretName, tokenName, key, secret, token string) error {
+	if key == "" && secret == "" && token != "" {
+		return fmt.Errorf("%w: %s requires %s and %s", ErrIncompleteStaticCredentials, tokenName, keyName, secretName)
+	}
+	if (key == "") != (secret == "") {
+		return fmt.Errorf("%w: %s and %s must be configured together", ErrIncompleteStaticCredentials, keyName, secretName)
+	}
+
+	return nil
+}

io/gocloud/s3.go

@@ -28,6 +28,7 @@ import (
 	"strconv"
 	"time"
 
+	internalaws "github.com/apache/iceberg-go/internal/awsconfig"
 	"github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/utils"
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -68,9 +69,12 @@ func ParseAWSConfig(ctx context.Context, props map[string]string) (*aws.Config,
 
 	accessKey, secretAccessKey := props[io.S3AccessKeyID], props[io.S3SecretAccessKey]
 	token := props[io.S3SessionToken]
-	if accessKey != "" || secretAccessKey != "" || token != "" {
+	if err := internalaws.ValidateStaticCredentials(io.S3AccessKeyID, io.S3SecretAccessKey, io.S3SessionToken, accessKey, secretAccessKey, token); err != nil {
+		return nil, err
+	}
+	if accessKey != "" {
 		opts = append(opts, config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
-			props[io.S3AccessKeyID], props[io.S3SecretAccessKey], props[io.S3SessionToken],
+			accessKey, secretAccessKey, token,
 		)))
 	}
 
@@ -174,6 +178,13 @@ func resolveUsePathStyle(endpoint string, props map[string]string) bool {
 // resolveS3AWSConfig returns the AWS config for the S3 FileIO, preferring an
 // ambient context config but letting explicit s3.* credentials override it.
 func resolveS3AWSConfig(ctx context.Context, props map[string]string) (*aws.Config, error) {
+	if err := internalaws.ValidateStaticCredentials(
+		io.S3AccessKeyID, io.S3SecretAccessKey, io.S3SessionToken,
+		props[io.S3AccessKeyID], props[io.S3SecretAccessKey], props[io.S3SessionToken],
+	); err != nil {
+		return nil, err
+	}
+
 	var (
 		base *aws.Config
 		err  error
@@ -195,9 +206,7 @@ func resolveS3AWSConfig(ctx context.Context, props map[string]string) (*aws.Conf
 		cfg.Region = r
 	}
 
-	// A complete explicit key pair overrides the credentials. A partial set
-	// (missing the access key or secret) falls through to the context/default
-	// chain rather than installing a provider with blank fields.
+	// A complete explicit key pair overrides the credentials.
 	if props[io.S3AccessKeyID] != "" && props[io.S3SecretAccessKey] != "" {
 		cfg.Credentials = credentials.NewStaticCredentialsProvider(
 			props[io.S3AccessKeyID], props[io.S3SecretAccessKey], props[io.S3SessionToken],

io/gocloud/s3_test.go

@@ -26,6 +26,7 @@ import (
 	"testing"
 	"time"
 
+	internalaws "github.com/apache/iceberg-go/internal/awsconfig"
 	"github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/utils"
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -95,12 +96,12 @@ func TestResolveS3AWSConfigCredentialPrecedence(t *testing.T) {
 		assert.Nil(t, shared.HTTPClient, "shared ctx config must stay unmutated")
 	})
 
-	// A partial key set (missing the secret) must not clobber the context creds.
-	t.Run("partial props creds fall through", func(t *testing.T) {
+	// A partial key set must fail consistently, even with ambient credentials.
+	t.Run("partial props creds are rejected", func(t *testing.T) {
 		t.Parallel()
-		cfg, err := resolveS3AWSConfig(ctxWith, map[string]string{io.S3AccessKeyID: "PARTIAL"})
-		require.NoError(t, err)
-		assert.Equal(t, "CTX", retrieve(t, cfg), "incomplete override must keep ctx creds")
+		_, err := resolveS3AWSConfig(ctxWith, map[string]string{io.S3AccessKeyID: "PARTIAL"})
+		require.ErrorIs(t, err, internalaws.ErrIncompleteStaticCredentials)
+		require.ErrorContains(t, err, "s3.access-key-id and s3.secret-access-key must be configured together")
 	})
 
 	// Explicit region overrides the context config's region, without mutating it.
@@ -114,6 +115,60 @@ func TestResolveS3AWSConfigCredentialPrecedence(t *testing.T) {
 	})
 }
 
+func TestParseAWSConfigRejectsIncompleteStaticCredentials(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name  string
+		props map[string]string
+		err   string
+	}{
+		{"access key only", map[string]string{io.S3AccessKeyID: "access"}, "s3.access-key-id and s3.secret-access-key must be configured together"},
+		{"secret key only", map[string]string{io.S3SecretAccessKey: "secret"}, "s3.access-key-id and s3.secret-access-key must be configured together"},
+		{"session token only", map[string]string{io.S3SessionToken: "token"}, "s3.session-token requires s3.access-key-id and s3.secret-access-key"},
+		{"access key and token", map[string]string{io.S3AccessKeyID: "access", io.S3SessionToken: "token"}, "s3.access-key-id and s3.secret-access-key must be configured together"},
+		{"secret key and token", map[string]string{io.S3SecretAccessKey: "secret", io.S3SessionToken: "token"}, "s3.access-key-id and s3.secret-access-key must be configured together"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			_, err := ParseAWSConfig(context.Background(), tt.props)
+			require.ErrorIs(t, err, internalaws.ErrIncompleteStaticCredentials)
+			require.ErrorContains(t, err, tt.err)
+		})
+	}
+
+	t.Run("empty props use the default chain", func(t *testing.T) {
+		t.Parallel()
+		cfg, err := ParseAWSConfig(context.Background(), map[string]string{})
+		require.NoError(t, err)
+		_, static := cfg.Credentials.(credentials.StaticCredentialsProvider)
+		assert.False(t, static)
+	})
+
+	t.Run("key pair without token is accepted", func(t *testing.T) {
+		t.Parallel()
+		cfg, err := ParseAWSConfig(context.Background(), map[string]string{
+			io.S3AccessKeyID: "access", io.S3SecretAccessKey: "secret",
+		})
+		require.NoError(t, err)
+		creds, err := cfg.Credentials.Retrieve(context.Background())
+		require.NoError(t, err)
+		assert.Empty(t, creds.SessionToken)
+	})
+
+	cfg, err := ParseAWSConfig(context.Background(), map[string]string{
+		io.S3AccessKeyID: "access", io.S3SecretAccessKey: "secret", io.S3SessionToken: "token",
+	})
+	require.NoError(t, err)
+	creds, err := cfg.Credentials.Retrieve(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, "access", creds.AccessKeyID)
+	require.Equal(t, "secret", creds.SecretAccessKey)
+	require.Equal(t, "token", creds.SessionToken)
+}
+
 func TestParseAWSConfigRemoteSigningEnabled(t *testing.T) {
 	t.Parallel()