Add support for AWS STS endpoint in the Bucket API

Signed-off-by: Matheus Pimenta <matheuscscp@gmail.com>
2024-07-19 13:15:25 +01:00 · 2024-07-19 13:15:25 +01:00 · 7536ab4b02
parent 218af573a3
commit 7536ab4b02
13 changed files with 714 additions and 48 deletions
--- a/api/v1/condition_types.go
+++ b/api/v1/condition_types.go
@ -108,4 +108,7 @@ const (
 	// PatchOperationFailedReason signals a failure in patching a kubernetes API
 	// object.
 	PatchOperationFailedReason string = "PatchOperationFailed"
+
+	// InvalidSTSConfigurationReason signals that the STS configurtion is invalid.
+	InvalidSTSConfigurationReason string = "InvalidSTSConfiguration"
 )
--- a/api/v1beta2/bucket_types.go
+++ b/api/v1beta2/bucket_types.go
@ -49,6 +49,8 @@ const (

 // BucketSpec specifies the required configuration to produce an Artifact for
 // an object storage bucket.
+// +kubebuilder:validation:XValidation:rule="self.provider == 'aws' || !has(self.sts)", message="STS configuration is only supported for the 'aws' Bucket provider"
+// +kubebuilder:validation:XValidation:rule="self.provider != 'aws' || !has(self.sts) || self.sts.provider == 'aws'", message="'aws' is the only supported STS provider for the 'aws' Bucket provider"
 type BucketSpec struct {
 	// Provider of the object storage bucket.
 	// Defaults to 'generic', which expects an S3 (API) compatible object
@ -66,6 +68,14 @@ type BucketSpec struct {
 	// +required
 	Endpoint string `json:"endpoint"`

+	// STS specifies the required configuration to use a Security Token
+	// Service for fetching temporary credentials to authenticate in a
+	// Bucket provider.
+	//
+	// This field is only supported for the `aws` provider.
+	// +optional
+	STS *BucketSTSSpec `json:"sts,omitempty"`
+
 	// Insecure allows connecting to a non-TLS HTTP Endpoint.
 	// +optional
 	Insecure bool `json:"insecure,omitempty"`
@ -140,6 +150,22 @@ type BucketSpec struct {
 	AccessFrom *acl.AccessFrom `json:"accessFrom,omitempty"`
 }

+// BucketSTSSpec specifies the required configuration to use a Security Token
+// Service for fetching temporary credentials to authenticate in a Bucket
+// provider.
+type BucketSTSSpec struct {
+	// Provider of the Security Token Service.
+	// +kubebuilder:validation:Enum=aws
+	// +required
+	Provider string `json:"provider"`
+
+	// Endpoint is the HTTP/S endpoint of the Security Token Service from
+	// where temporary credentials will be fetched.
+	// +required
+	// +kubebuilder:validation:Pattern="^(http|https)://.*$"
+	Endpoint string `json:"endpoint"`
+}
+
 // BucketStatus records the observed state of a Bucket.
 type BucketStatus struct {
 	// ObservedGeneration is the last observed generation of the Bucket object.
--- a/api/v1beta2/sts_types.go
+++ b/api/v1beta2/sts_types.go
@ -0,0 +1,23 @@
+/*
+Copyright 2024 The Flux authors
+
+Licensed 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 v1beta2
+
+const (
+	// STSProviderAmazon represents the AWS provider for Security Token Service.
+	// Provides support for fetching temporary credentials from an AWS STS endpoint.
+	STSProviderAmazon string = "aws"
+)
--- a/api/v1beta2/zz_generated.deepcopy.go
+++ b/api/v1beta2/zz_generated.deepcopy.go
@ -115,9 +115,29 @@ func (in *BucketList) DeepCopyObject() runtime.Object {
 	return nil
 }

+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *BucketSTSSpec) DeepCopyInto(out *BucketSTSSpec) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BucketSTSSpec.
+func (in *BucketSTSSpec) DeepCopy() *BucketSTSSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(BucketSTSSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *BucketSpec) DeepCopyInto(out *BucketSpec) {
 	*out = *in
+	if in.STS != nil {
+		in, out := &in.STS, &out.STS
+		*out = new(BucketSTSSpec)
+		**out = **in
+	}
 	if in.SecretRef != nil {
 		in, out := &in.SecretRef, &out.SecretRef
 		*out = new(meta.LocalObjectReference)
--- a/config/crd/bases/source.toolkit.fluxcd.io_buckets.yaml
+++ b/config/crd/bases/source.toolkit.fluxcd.io_buckets.yaml
@ -420,6 +420,30 @@ spec:
                required:
                - name
                type: object
+              sts:
+                description: |-
+                  STS specifies the required configuration to use a Security Token
+                  Service for fetching temporary credentials to authenticate in a
+                  Bucket provider.
+
+
+                  This field is only supported for the `aws` provider.
+                properties:
+                  endpoint:
+                    description: |-
+                      Endpoint is the HTTP/S endpoint of the Security Token Service from
+                      where temporary credentials will be fetched.
+                    pattern: ^(http|https)://.*$
+                    type: string
+                  provider:
+                    description: Provider of the Security Token Service.
+                    enum:
+                    - aws
+                    type: string
+                required:
+                - endpoint
+                - provider
+                type: object
              suspend:
                description: |-
                  Suspend tells the controller to suspend the reconciliation of this
@ -435,6 +459,13 @@ spec:
            - endpoint
            - interval
            type: object
+            x-kubernetes-validations:
+            - message: STS configuration is only supported for the 'aws' Bucket provider
+              rule: self.provider == 'aws' || !has(self.sts)
+            - message: '''aws'' is the only supported STS provider for the ''aws''
+                Bucket provider'
+              rule: self.provider != 'aws' || !has(self.sts) || self.sts.provider
+                == 'aws'
          status:
            default:
              observedGeneration: -1
--- a/docs/api/v1beta2/source.md
+++ b/docs/api/v1beta2/source.md
@ -114,6 +114,23 @@ string
 </tr>
 <tr>
 <td>
+<code>sts</code><br>
+<em>
+<a href="#source.toolkit.fluxcd.io/v1beta2.BucketSTSSpec">
+BucketSTSSpec
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>STS specifies the required configuration to use a Security Token
+Service for fetching temporary credentials to authenticate in a
+Bucket provider.</p>
+<p>This field is only supported for the <code>aws</code> provider.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>insecure</code><br>
 <em>
 bool
@ -1424,6 +1441,52 @@ map[string]string
 </table>
 </div>
 </div>
+<h3 id="source.toolkit.fluxcd.io/v1beta2.BucketSTSSpec">BucketSTSSpec
+</h3>
+<p>
+(<em>Appears on:</em>
+<a href="#source.toolkit.fluxcd.io/v1beta2.BucketSpec">BucketSpec</a>)
+</p>
+<p>BucketSTSSpec specifies the required configuration to use a Security Token
+Service for fetching temporary credentials to authenticate in a Bucket
+provider.</p>
+<div class="md-typeset__scrollwrap">
+<div class="md-typeset__table">
+<table>
+<thead>
+<tr>
+<th>Field</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>provider</code><br>
+<em>
+string
+</em>
+</td>
+<td>
+<p>Provider of the Security Token Service.</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>endpoint</code><br>
+<em>
+string
+</em>
+</td>
+<td>
+<p>Endpoint is the HTTP/S endpoint of the Security Token Service from
+where temporary credentials will be fetched.</p>
+</td>
+</tr>
+</tbody>
+</table>
+</div>
+</div>
 <h3 id="source.toolkit.fluxcd.io/v1beta2.BucketSpec">BucketSpec
 </h3>
 <p>
@ -1480,6 +1543,23 @@ string
 </tr>
 <tr>
 <td>
+<code>sts</code><br>
+<em>
+<a href="#source.toolkit.fluxcd.io/v1beta2.BucketSTSSpec">
+BucketSTSSpec
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>STS specifies the required configuration to use a Security Token
+Service for fetching temporary credentials to authenticate in a
+Bucket provider.</p>
+<p>This field is only supported for the <code>aws</code> provider.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>insecure</code><br>
 <em>
 bool
--- a/docs/spec/v1beta2/buckets.md
+++ b/docs/spec/v1beta2/buckets.md
@ -749,6 +749,23 @@ HTTP endpoint requires enabling [`.spec.insecure`](#insecure).
 Some endpoints require the specification of a [`.spec.region`](#region),
 see [Provider](#provider) for more (provider specific) examples.

+### STS
+
+`.spec.sts` is an optional field for specifying the Security Token Service
+configuration. A Security Token Service (STS) is a web service that issues
+temporary security credentials. By adding this field, one may specify the
+STS endpoint from where temporary credentials will be fetched.
+
+If using `.spec.sts`, the following fields are required:
+
+- `.spec.sts.provider`, the Security Token Service provider. The only supported
+  option is `aws`.
+- `.spec.sts.endpoint`, the HTTP/S endpoint of the Security Token Service. In
+  the case of AWS, this can be `https://sts.amazonaws.com`, or a Regional STS
+  Endpoint, or an Interface Endpoint created inside a VPC.
+
+This field is only supported for the `aws` bucket provider.
+
 ### Bucket name

 `.spec.bucketName` is a required field that specifies which object storage
--- a/internal/controller/bucket_controller.go
+++ b/internal/controller/bucket_controller.go
@ -463,6 +463,19 @@ func (r *BucketReconciler) reconcileSource(ctx context.Context, sp *patch.Serial
 			conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, e.Reason, "%s", e)
 			return sreconcile.ResultEmpty, e
 		}
+		if sts := obj.Spec.STS; sts != nil {
+			if err := minio.ValidateSTSProvider(obj.Spec.Provider, sts.Provider); err != nil {
+				e := serror.NewStalling(err, sourcev1.InvalidSTSConfigurationReason)
+				conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, e.Reason, "%s", e)
+				return sreconcile.ResultEmpty, e
+			}
+			if _, err := url.Parse(sts.Endpoint); err != nil {
+				err := fmt.Errorf("failed to parse STS endpoint '%s': %w", sts.Endpoint, err)
+				e := serror.NewStalling(err, sourcev1.URLInvalidReason)
+				conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, e.Reason, "%s", e)
+				return sreconcile.ResultEmpty, e
+			}
+		}
 		tlsConfig, err := r.getTLSConfig(ctx, obj)
 		if err != nil {
 			e := serror.NewGeneric(err, sourcev1.AuthenticationFailedReason)
--- a/internal/controller/bucket_controller_test.go
+++ b/internal/controller/bucket_controller_test.go
@ -608,6 +608,45 @@ func TestBucketReconciler_reconcileSource_generic(t *testing.T) {
 				*conditions.UnknownCondition(meta.ReadyCondition, "foo", "bar"),
 			},
 		},
+		{
+			name:       "Observes incompatible STS provider",
+			bucketName: "dummy",
+			beforeFunc: func(obj *bucketv1.Bucket) {
+				obj.Spec.Provider = "generic"
+				obj.Spec.STS = &bucketv1.BucketSTSSpec{
+					Provider: "aws",
+				}
+				conditions.MarkReconciling(obj, meta.ProgressingReason, "foo")
+				conditions.MarkUnknown(obj, meta.ReadyCondition, "foo", "bar")
+			},
+			wantErr:     true,
+			assertIndex: index.NewDigester(),
+			assertConditions: []metav1.Condition{
+				*conditions.TrueCondition(sourcev1.FetchFailedCondition, sourcev1.InvalidSTSConfigurationReason, "STS configuration is not supported for 'generic' bucket provider"),
+				*conditions.TrueCondition(meta.ReconcilingCondition, meta.ProgressingReason, "foo"),
+				*conditions.UnknownCondition(meta.ReadyCondition, "foo", "bar"),
+			},
+		},
+		{
+			name:       "Observes invalid STS endpoint",
+			bucketName: "dummy",
+			beforeFunc: func(obj *bucketv1.Bucket) {
+				obj.Spec.Provider = "aws" // TODO: change to generic when ldap STS provider is implemented
+				obj.Spec.STS = &bucketv1.BucketSTSSpec{
+					Provider: "aws", // TODO: change to ldap when ldap STS provider is implemented
+					Endpoint: "something\t",
+				}
+				conditions.MarkReconciling(obj, meta.ProgressingReason, "foo")
+				conditions.MarkUnknown(obj, meta.ReadyCondition, "foo", "bar")
+			},
+			wantErr:     true,
+			assertIndex: index.NewDigester(),
+			assertConditions: []metav1.Condition{
+				*conditions.TrueCondition(sourcev1.FetchFailedCondition, sourcev1.URLInvalidReason, "failed to parse STS endpoint 'something\t': parse \"something\\t\": net/url: invalid control character in URL"),
+				*conditions.TrueCondition(meta.ReconcilingCondition, meta.ProgressingReason, "foo"),
+				*conditions.UnknownCondition(meta.ReadyCondition, "foo", "bar"),
+			},
+		},
 		{
 			name: "Transient bucket name API failure",
 			beforeFunc: func(obj *bucketv1.Bucket) {
@ -1762,3 +1801,119 @@ func TestBucketReconciler_getProxyURL(t *testing.T) {
 		})
 	}
 }
+
+func TestBucketReconciler_APIServerValidation_STS(t *testing.T) {
+	tests := []struct {
+		name           string
+		bucketProvider string
+		stsConfig      *bucketv1.BucketSTSSpec
+		err            string
+	}{
+		{
+			name:           "gcp unsupported",
+			bucketProvider: "gcp",
+			stsConfig: &bucketv1.BucketSTSSpec{
+				Provider: "aws",
+				Endpoint: "http://test",
+			},
+			err: "STS configuration is only supported for the 'aws' Bucket provider",
+		},
+		{
+			name:           "azure unsupported",
+			bucketProvider: "azure",
+			stsConfig: &bucketv1.BucketSTSSpec{
+				Provider: "aws",
+				Endpoint: "http://test",
+			},
+			err: "STS configuration is only supported for the 'aws' Bucket provider",
+		},
+		{
+			name:           "generic unsupported",
+			bucketProvider: "generic",
+			stsConfig: &bucketv1.BucketSTSSpec{
+				Provider: "aws",
+				Endpoint: "http://test",
+			},
+			err: "STS configuration is only supported for the 'aws' Bucket provider",
+		},
+		{
+			name:           "aws supported",
+			bucketProvider: "aws",
+			stsConfig: &bucketv1.BucketSTSSpec{
+				Provider: "aws",
+				Endpoint: "http://test",
+			},
+		},
+		{
+			name:           "invalid endpoint",
+			bucketProvider: "aws",
+			stsConfig: &bucketv1.BucketSTSSpec{
+				Provider: "aws",
+				Endpoint: "test",
+			},
+			err: "spec.sts.endpoint in body should match '^(http|https)://.*$'",
+		},
+		{
+			name:           "gcp can be created without STS config",
+			bucketProvider: "gcp",
+		},
+		{
+			name:           "azure can be created without STS config",
+			bucketProvider: "azure",
+		},
+		{
+			name:           "generic can be created without STS config",
+			bucketProvider: "generic",
+		},
+		{
+			name:           "aws can be created without STS config",
+			bucketProvider: "aws",
+		},
+		// Can't be tested at present with only one allowed sts provider.
+		// {
+		// 	name:           "ldap unsupported for aws",
+		// 	bucketProvider: "aws",
+		// 	stsConfig: &bucketv1.BucketSTSSpec{
+		// 		Provider: "ldap",
+		// 		Endpoint: "http://test",
+		// 	},
+		// 	err: "'aws' is the only supported STS provider for the 'aws' Bucket provider",
+		// },
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			obj := &bucketv1.Bucket{
+				ObjectMeta: metav1.ObjectMeta{
+					GenerateName: "bucket-reconcile-",
+					Namespace:    "default",
+				},
+				Spec: bucketv1.BucketSpec{
+					Provider:   tt.bucketProvider,
+					BucketName: "test",
+					Endpoint:   "test",
+					Suspend:    true,
+					Interval:   metav1.Duration{Duration: interval},
+					Timeout:    &metav1.Duration{Duration: timeout},
+					STS:        tt.stsConfig,
+				},
+			}
+
+			err := testEnv.Create(ctx, obj)
+			if err == nil {
+				defer func() {
+					err := testEnv.Delete(ctx, obj)
+					g.Expect(err).NotTo(HaveOccurred())
+				}()
+			}
+
+			if tt.err != "" {
+				g.Expect(err.Error()).To(ContainSubstring(tt.err))
+			} else {
+				g.Expect(err).NotTo(HaveOccurred())
+			}
+		})
+	}
+}
--- a/pkg/minio/minio.go
+++ b/pkg/minio/minio.go
@ -71,14 +71,10 @@ func WithProxyURL(proxyURL *url.URL) Option {

 // NewClient creates a new Minio storage client.
 func NewClient(bucket *sourcev1.Bucket, opts ...Option) (*MinioClient, error) {
-
 	var o options
 	for _, opt := range opts {
 		opt(&o)
 	}
-	secret := o.secret
-	tlsConfig := o.tlsConfig
-	proxyURL := o.proxyURL

 	minioOpts := minio.Options{
 		Region: bucket.Spec.Region,
@ -88,32 +84,24 @@ func NewClient(bucket *sourcev1.Bucket, opts ...Option) (*MinioClient, error) {
 		// auto access, which we believe can cover most use cases.
 	}

-	if secret != nil {
-		var accessKey, secretKey string
-		if k, ok := secret.Data["accesskey"]; ok {
-			accessKey = string(k)
-		}
-		if k, ok := secret.Data["secretkey"]; ok {
-			secretKey = string(k)
-		}
-		if accessKey != "" && secretKey != "" {
-			minioOpts.Creds = credentials.NewStaticV4(accessKey, secretKey, "")
-		}
-	} else if bucket.Spec.Provider == sourcev1.AmazonBucketProvider {
-		minioOpts.Creds = credentials.NewIAM("")
+	switch bucketProvider := bucket.Spec.Provider; {
+	case o.secret != nil:
+		minioOpts.Creds = newCredsFromSecret(o.secret)
+	case bucketProvider == sourcev1.AmazonBucketProvider:
+		minioOpts.Creds = newAWSCreds(bucket, o.proxyURL)
 	}

 	var transportOpts []func(*http.Transport)

-	if minioOpts.Secure && tlsConfig != nil {
+	if minioOpts.Secure && o.tlsConfig != nil {
 		transportOpts = append(transportOpts, func(t *http.Transport) {
-			t.TLSClientConfig = tlsConfig.Clone()
+			t.TLSClientConfig = o.tlsConfig.Clone()
 		})
 	}

-	if proxyURL != nil {
+	if o.proxyURL != nil {
 		transportOpts = append(transportOpts, func(t *http.Transport) {
-			t.Proxy = http.ProxyURL(proxyURL)
+			t.Proxy = http.ProxyURL(o.proxyURL)
 		})
 	}

@ -135,6 +123,42 @@ func NewClient(bucket *sourcev1.Bucket, opts ...Option) (*MinioClient, error) {
 	return &MinioClient{Client: client}, nil
 }

+// newCredsFromSecret creates a new Minio credentials object from the provided
+// secret.
+func newCredsFromSecret(secret *corev1.Secret) *credentials.Credentials {
+	var accessKey, secretKey string
+	if k, ok := secret.Data["accesskey"]; ok {
+		accessKey = string(k)
+	}
+	if k, ok := secret.Data["secretkey"]; ok {
+		secretKey = string(k)
+	}
+	if accessKey != "" && secretKey != "" {
+		return credentials.NewStaticV4(accessKey, secretKey, "")
+	}
+	return nil
+}
+
+// newAWSCreds creates a new Minio credentials object for `aws` bucket provider.
+func newAWSCreds(bucket *sourcev1.Bucket, proxyURL *url.URL) *credentials.Credentials {
+	stsEndpoint := ""
+	if sts := bucket.Spec.STS; sts != nil {
+		stsEndpoint = sts.Endpoint
+	}
+
+	creds := credentials.NewIAM(stsEndpoint)
+	if proxyURL != nil {
+		transport := http.DefaultTransport.(*http.Transport).Clone()
+		transport.Proxy = http.ProxyURL(proxyURL)
+		client := &http.Client{Transport: transport}
+		creds = credentials.New(&credentials.IAM{
+			Client:   client,
+			Endpoint: stsEndpoint,
+		})
+	}
+	return creds
+}
+
 // ValidateSecret validates the credential secret. The provided Secret may
 // be nil.
 func ValidateSecret(secret *corev1.Secret) error {
@ -151,6 +175,24 @@ func ValidateSecret(secret *corev1.Secret) error {
 	return nil
 }

+// ValidateSTSProvider validates the STS provider.
+func ValidateSTSProvider(bucketProvider, stsProvider string) error {
+	errProviderIncompatbility := fmt.Errorf("STS provider '%s' is not supported for '%s' bucket provider",
+		stsProvider, bucketProvider)
+
+	switch bucketProvider {
+	case sourcev1.AmazonBucketProvider:
+		switch stsProvider {
+		case sourcev1.STSProviderAmazon:
+			return nil
+		default:
+			return errProviderIncompatbility
+		}
+	}
+
+	return fmt.Errorf("STS configuration is not supported for '%s' bucket provider", bucketProvider)
+}
+
 // FGetObject gets the object from the provided object storage bucket, and
 // writes it to targetPath.
 // It returns the etag of the successfully fetched file, or any error.
--- a/pkg/minio/minio_test.go
+++ b/pkg/minio/minio_test.go
@ -20,10 +20,10 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"log"
-	"net"
 	"net/http"
 	"net/url"
 	"os"
@ -32,9 +32,9 @@ import (
 	"testing"
 	"time"

-	"github.com/elazarl/goproxy"
 	"github.com/google/uuid"
 	miniov7 "github.com/minio/minio-go/v7"
+	"github.com/minio/minio-go/v7/pkg/credentials"
 	"github.com/ory/dockertest/v3"
 	"github.com/ory/dockertest/v3/docker"
 	"gotest.tools/assert"
@ -45,6 +45,8 @@ import (
 	"github.com/fluxcd/pkg/sourceignore"

 	sourcev1 "github.com/fluxcd/source-controller/api/v1beta2"
+	testlistener "github.com/fluxcd/source-controller/tests/listener"
+	testproxy "github.com/fluxcd/source-controller/tests/proxy"
 )

 const (
@ -244,34 +246,153 @@ func TestFGetObject(t *testing.T) {
 	assert.NilError(t, err)
 }

-func TestNewClientAndFGetObjectWithProxy(t *testing.T) {
-	// start proxy
-	proxyListener, err := net.Listen("tcp", ":0")
-	assert.NilError(t, err, "could not start proxy server")
-	defer proxyListener.Close()
-	proxyAddr := proxyListener.Addr().String()
-	proxyHandler := goproxy.NewProxyHttpServer()
-	proxyHandler.Verbose = true
-	proxyServer := &http.Server{
-		Addr:    proxyAddr,
-		Handler: proxyHandler,
+func TestNewClientAndFGetObjectWithSTSEndpoint(t *testing.T) {
+	// start a mock STS server
+	stsListener, stsAddr, stsPort := testlistener.New(t)
+	stsEndpoint := fmt.Sprintf("http://%s", stsAddr)
+	stsHandler := http.NewServeMux()
+	stsHandler.HandleFunc("PUT "+credentials.TokenPath,
+		func(w http.ResponseWriter, r *http.Request) {
+			_, err := w.Write([]byte("mock-token"))
+			assert.NilError(t, err)
+		})
+	stsHandler.HandleFunc("GET "+credentials.DefaultIAMSecurityCredsPath,
+		func(w http.ResponseWriter, r *http.Request) {
+			token := r.Header.Get(credentials.TokenRequestHeader)
+			assert.Equal(t, token, "mock-token")
+			_, err := w.Write([]byte("mock-role"))
+			assert.NilError(t, err)
+		})
+	var roleCredsRetrieved bool
+	stsHandler.HandleFunc("GET "+credentials.DefaultIAMSecurityCredsPath+"mock-role",
+		func(w http.ResponseWriter, r *http.Request) {
+			token := r.Header.Get(credentials.TokenRequestHeader)
+			assert.Equal(t, token, "mock-token")
+			err := json.NewEncoder(w).Encode(map[string]any{
+				"Code":            "Success",
+				"AccessKeyID":     testMinioRootUser,
+				"SecretAccessKey": testMinioRootPassword,
+			})
+			assert.NilError(t, err)
+			roleCredsRetrieved = true
+		})
+	stsServer := &http.Server{
+		Addr:    stsAddr,
+		Handler: stsHandler,
+	}
+	go stsServer.Serve(stsListener)
+	defer stsServer.Shutdown(context.Background())
+
+	// start proxy
+	proxyAddr, proxyPort := testproxy.New(t)
+
+	tests := []struct {
+		name     string
+		provider string
+		stsSpec  *sourcev1.BucketSTSSpec
+		opts     []Option
+		err      string
+	}{
+		{
+			name:     "with correct endpoint",
+			provider: "aws",
+			stsSpec: &sourcev1.BucketSTSSpec{
+				Provider: "aws",
+				Endpoint: stsEndpoint,
+			},
+		},
+		{
+			name:     "with incorrect endpoint",
+			provider: "aws",
+			stsSpec: &sourcev1.BucketSTSSpec{
+				Provider: "aws",
+				Endpoint: fmt.Sprintf("http://localhost:%d", stsPort+1),
+			},
+			err: "connection refused",
+		},
+		{
+			name:     "with correct endpoint and proxy",
+			provider: "aws",
+			stsSpec: &sourcev1.BucketSTSSpec{
+				Provider: "aws",
+				Endpoint: stsEndpoint,
+			},
+			opts: []Option{WithProxyURL(&url.URL{Scheme: "http", Host: proxyAddr})},
+		},
+		{
+			name:     "with correct endpoint and incorrect proxy",
+			provider: "aws",
+			stsSpec: &sourcev1.BucketSTSSpec{
+				Provider: "aws",
+				Endpoint: stsEndpoint,
+			},
+			opts: []Option{WithProxyURL(&url.URL{Scheme: "http", Host: fmt.Sprintf("localhost:%d", proxyPort+1)})},
+			err:  "connection refused",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			roleCredsRetrieved = false
+			bucket := bucketStub(bucket, testMinioAddress)
+			bucket.Spec.Provider = tt.provider
+			bucket.Spec.STS = tt.stsSpec
+			minioClient, err := NewClient(bucket, append(tt.opts, WithTLSConfig(testTLSConfig))...)
+			assert.NilError(t, err)
+			assert.Assert(t, minioClient != nil)
+			ctx := context.Background()
+			tempDir := t.TempDir()
+			path := filepath.Join(tempDir, sourceignore.IgnoreFile)
+			_, err = minioClient.FGetObject(ctx, bucketName, objectName, path)
+			if tt.err != "" {
+				assert.ErrorContains(t, err, tt.err)
+			} else {
+				assert.NilError(t, err)
+				assert.Assert(t, roleCredsRetrieved)
+			}
+		})
+	}
+}
+
+func TestNewClientAndFGetObjectWithProxy(t *testing.T) {
+	proxyAddr, proxyPort := testproxy.New(t)
+
+	tests := []struct {
+		name         string
+		proxyURL     *url.URL
+		errSubstring string
+	}{
+		{
+			name:     "with correct proxy",
+			proxyURL: &url.URL{Scheme: "http", Host: proxyAddr},
+		},
+		{
+			name:         "with incorrect proxy",
+			proxyURL:     &url.URL{Scheme: "http", Host: fmt.Sprintf("localhost:%d", proxyPort+1)},
+			errSubstring: "connection refused",
+		},
 	}
-	go proxyServer.Serve(proxyListener)
-	defer proxyServer.Shutdown(context.Background())
-	proxyURL := &url.URL{Scheme: "http", Host: proxyAddr}

 	// run test
-	minioClient, err := NewClient(bucketStub(bucket, testMinioAddress),
-		WithSecret(secret.DeepCopy()),
-		WithTLSConfig(testTLSConfig),
-		WithProxyURL(proxyURL))
-	assert.NilError(t, err)
-	assert.Assert(t, minioClient != nil)
-	ctx := context.Background()
-	tempDir := t.TempDir()
-	path := filepath.Join(tempDir, sourceignore.IgnoreFile)
-	_, err = minioClient.FGetObject(ctx, bucketName, objectName, path)
-	assert.NilError(t, err)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			minioClient, err := NewClient(bucketStub(bucket, testMinioAddress),
+				WithSecret(secret.DeepCopy()),
+				WithTLSConfig(testTLSConfig),
+				WithProxyURL(tt.proxyURL))
+			assert.NilError(t, err)
+			assert.Assert(t, minioClient != nil)
+			ctx := context.Background()
+			tempDir := t.TempDir()
+			path := filepath.Join(tempDir, sourceignore.IgnoreFile)
+			_, err = minioClient.FGetObject(ctx, bucketName, objectName, path)
+			if tt.errSubstring != "" {
+				assert.ErrorContains(t, err, tt.errSubstring)
+			} else {
+				assert.NilError(t, err)
+			}
+		})
+	}
 }

 func TestFGetObjectNotExists(t *testing.T) {
@ -349,6 +470,47 @@ func TestValidateSecret(t *testing.T) {
 	}
 }

+func TestValidateSTSProvider(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name           string
+		bucketProvider string
+		stsProvider    string
+		err            string
+	}{
+		{
+			name:           "aws",
+			bucketProvider: "aws",
+			stsProvider:    "aws",
+		},
+		{
+			name:           "unsupported for aws",
+			bucketProvider: "aws",
+			stsProvider:    "ldap",
+			err:            "STS provider 'ldap' is not supported for 'aws' bucket provider",
+		},
+		{
+			name:           "unsupported bucket provider",
+			bucketProvider: "gcp",
+			stsProvider:    "gcp",
+			err:            "STS configuration is not supported for 'gcp' bucket provider",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			err := ValidateSTSProvider(tt.bucketProvider, tt.stsProvider)
+			if tt.err != "" {
+				assert.Error(t, err, tt.err)
+			} else {
+				assert.NilError(t, err)
+			}
+		})
+	}
+}
+
 func bucketStub(bucket sourcev1.Bucket, endpoint string) *sourcev1.Bucket {
 	b := bucket.DeepCopy()
 	b.Spec.Endpoint = endpoint
--- a/tests/listener/listener.go
+++ b/tests/listener/listener.go
@ -0,0 +1,46 @@
+/*
+Copyright 2024 The Flux authors
+
+Licensed 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 testlistener
+
+import (
+	"net"
+	"strconv"
+	"strings"
+	"testing"
+
+	"gotest.tools/assert"
+)
+
+// New creates a TCP listener on a random port and returns
+// the listener, the address and the port of this listener.
+// It also registers a cleanup function to close the listener
+// when the test ends.
+func New(t *testing.T) (net.Listener, string, int) {
+	t.Helper()
+
+	lis, err := net.Listen("tcp", ":0")
+	assert.NilError(t, err)
+	t.Cleanup(func() { lis.Close() })
+
+	addr := lis.Addr().String()
+	addrParts := strings.Split(addr, ":")
+	portStr := addrParts[len(addrParts)-1]
+	port, err := strconv.Atoi(portStr)
+	assert.NilError(t, err)
+
+	return lis, addr, port
+}
--- a/tests/proxy/proxy.go
+++ b/tests/proxy/proxy.go
@ -0,0 +1,48 @@
+/*
+Copyright 2024 The Flux authors
+
+Licensed 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 testproxy
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/elazarl/goproxy"
+
+	testlistener "github.com/fluxcd/source-controller/tests/listener"
+)
+
+// New creates a new goproxy server on a random port and returns
+// the address and the port of this server. It also registers a
+// cleanup functions to close the server and the listener when
+// the test ends.
+func New(t *testing.T) (string, int) {
+	t.Helper()
+
+	lis, addr, port := testlistener.New(t)
+
+	handler := goproxy.NewProxyHttpServer()
+	handler.Verbose = true
+
+	server := &http.Server{
+		Addr:    addr,
+		Handler: handler,
+	}
+	go server.Serve(lis)
+	t.Cleanup(func() { server.Close() })
+
+	return addr, port
+}