[RFC-0010] Add multi-tenant workload identity support for GCP Bucket

Signed-off-by: cappyzawa <cappyzawa@gmail.com>

api/v1/bucket_types.go

@@ -51,6 +51,8 @@ const (
 // +kubebuilder:validation:XValidation:rule="self.provider != 'generic' || !has(self.sts) || self.sts.provider == 'ldap'", message="'ldap' is the only supported STS provider for the 'generic' Bucket provider"
 // +kubebuilder:validation:XValidation:rule="!has(self.sts) || self.sts.provider != 'aws' || !has(self.sts.secretRef)", message="spec.sts.secretRef is not required for the 'aws' STS provider"
 // +kubebuilder:validation:XValidation:rule="!has(self.sts) || self.sts.provider != 'aws' || !has(self.sts.certSecretRef)", message="spec.sts.certSecretRef is not required for the 'aws' STS provider"
+// +kubebuilder:validation:XValidation:rule="self.provider == 'gcp' || !has(self.serviceAccountName)", message="ServiceAccountName is only supported for the 'gcp' Bucket provider"
+// +kubebuilder:validation:XValidation:rule="!has(self.secretRef) || !has(self.serviceAccountName)", message="cannot set both .spec.secretRef and .spec.serviceAccountName"
 type BucketSpec struct {
 	// Provider of the object storage bucket.
 	// Defaults to 'generic', which expects an S3 (API) compatible object
@@ -93,6 +95,12 @@ type BucketSpec struct {
 	// +optional
 	SecretRef *meta.LocalObjectReference `json:"secretRef,omitempty"`
 
+	// ServiceAccountName is the name of the Kubernetes ServiceAccount used to authenticate
+	// the bucket. For more information about workload identity:
+	// https://fluxcd.io/flux/components/source/buckets/#workload-identity
+	// +optional
+	ServiceAccountName string `json:"serviceAccountName,omitempty"`
+
 	// CertSecretRef can be given the name of a Secret containing
 	// either or both of
 	//

config/crd/bases/source.toolkit.fluxcd.io_buckets.yaml

@@ -142,6 +142,12 @@ spec:
                 required:
                 - name
                 type: object
+              serviceAccountName:
+                description: |-
+                  ServiceAccountName is the name of the Kubernetes ServiceAccount used to authenticate
+                  the bucket. For more information about workload identity:
+                  https://fluxcd.io/flux/components/source/buckets/#workload-identity
+                type: string
               sts:
                 description: |-
                   STS specifies the required configuration to use a Security Token
@@ -232,6 +238,10 @@ spec:
               rule: '!has(self.sts) || self.sts.provider != ''aws'' || !has(self.sts.secretRef)'
             - message: spec.sts.certSecretRef is not required for the 'aws' STS provider
               rule: '!has(self.sts) || self.sts.provider != ''aws'' || !has(self.sts.certSecretRef)'
+            - message: ServiceAccountName is only supported for the 'gcp' Bucket provider
+              rule: self.provider == 'gcp' || !has(self.serviceAccountName)
+            - message: cannot set both .spec.secretRef and .spec.serviceAccountName
+              rule: '!has(self.secretRef) || !has(self.serviceAccountName)'
           status:
             default:
               observedGeneration: -1

config/rbac/role.yaml

@@ -15,6 +15,7 @@ rules:
   - ""
   resources:
   - secrets
+  - serviceaccounts
   verbs:
   - get
   - list

docs/api/v1/source.md

@@ -182,6 +182,20 @@ for the Bucket.</p>
 </tr>
 <tr>
 <td>
+<code>serviceAccountName</code><br>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>ServiceAccountName is the name of the Kubernetes ServiceAccount used to authenticate
+the bucket. For more information about workload identity:
+<a href="https://fluxcd.io/flux/components/source/buckets/#workload-identity">https://fluxcd.io/flux/components/source/buckets/#workload-identity</a></p>
+</td>
+</tr>
+<tr>
+<td>
 <code>certSecretRef</code><br>
 <em>
 <a href="https://pkg.go.dev/github.com/fluxcd/pkg/apis/meta#LocalObjectReference">
@@ -1624,6 +1638,20 @@ for the Bucket.</p>
 </tr>
 <tr>
 <td>
+<code>serviceAccountName</code><br>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>ServiceAccountName is the name of the Kubernetes ServiceAccount used to authenticate
+the bucket. For more information about workload identity:
+<a href="https://fluxcd.io/flux/components/source/buckets/#workload-identity">https://fluxcd.io/flux/components/source/buckets/#workload-identity</a></p>
+</td>
+</tr>
+<tr>
+<td>
 <code>certSecretRef</code><br>
 <em>
 <a href="https://pkg.go.dev/github.com/fluxcd/pkg/apis/meta#LocalObjectReference">

docs/spec/v1/buckets.md

@@ -647,29 +647,16 @@ Refer to the [Azure documentation](https://learn.microsoft.com/en-us/rest/api/st
 
 #### GCP
 
-When a Bucket's `.spec.provider` is set to `gcp`, the source-controller will
-attempt to communicate with the specified [Endpoint](#endpoint) using the
-[Google Client SDK](https://github.com/googleapis/google-api-go-client).
+For detailed setup instructions, see: https://fluxcd.io/flux/integrations/gcp/#for-google-cloud-storage
 
-Without a [Secret reference](#secret-reference), authorization using a
-workload identity is attempted by default. The workload identity is obtained
-using the `GOOGLE_APPLICATION_CREDENTIALS` environment variable, falling back
-to the Google Application Credential file in the config directory.
-When a reference is specified, it expects a Secret with a `.data.serviceaccount`
-value with a GCP service account JSON file.
-
-The Provider allows for specifying the
-[Bucket location](https://cloud.google.com/storage/docs/locations) using the
-[`.spec.region` field](#region).
-
-##### GCP example
+##### GCP Controller-Level Workload Identity example
 
 ```yaml
 ---
 apiVersion: source.toolkit.fluxcd.io/v1
 kind: Bucket
 metadata:
-  name: gcp-workload-identity
+  name: gcp-controller-level-workload-identity
   namespace: default
 spec:
   interval: 5m0s
@@ -680,6 +667,37 @@ spec:
   timeout: 30s
 ```
 
+##### GCP Object-Level Workload Identity example
+
+**Note:** To use Object-Level Workload Identity (`.spec.serviceAccountName` with 
+cloud providers), the controller feature gate `ObjectLevelWorkloadIdentity` must 
+be enabled.
+
+```yaml
+---
+apiVersion: source.toolkit.fluxcd.io/v1
+kind: Bucket
+metadata:
+  name: gcp-object-level-workload-identity
+  namespace: default
+spec:
+  interval: 5m0s
+  provider: gcp
+  bucketName: podinfo
+  endpoint: storage.googleapis.com
+  region: us-east-1
+  serviceAccountName: gcp-workload-identity-sa
+  timeout: 30s
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: gcp-workload-identity-sa
+  namespace: default
+  annotations:
+    iam.gke.io/gcp-service-account: <identity-name>
+```
+
 ##### GCP static auth example
 
 ```yaml
@@ -959,6 +977,29 @@ credentials for the object storage. For some `.spec.provider` implementations
 the presence of the field is required, see [Provider](#provider) for more
 details and examples.
 
+### Service Account reference
+
+`.spec.serviceAccountName` is an optional field to specify a Service Account
+in the same namespace as Bucket with purpose depending on the value of
+the `.spec.provider` field:
+
+- When `.spec.provider` is set to `generic`, the controller will fetch the image
+  pull secrets attached to the Service Account and use them for authentication.
+- When `.spec.provider` is set to `aws`, `azure`, or `gcp`, the Service Account
+  will be used for Workload Identity authentication. In this case, the controller
+  feature gate `ObjectLevelWorkloadIdentity` must be enabled, otherwise the
+  controller will error out.
+
+**Note:** that for a publicly accessible object storage, you don't need to
+provide a `secretRef` nor `serviceAccountName`.
+
+**Important:** `.spec.secretRef` and `.spec.serviceAccountName` are mutually
+exclusive and cannot be set at the same time. This constraint is enforced
+at the CRD level.
+
+For a complete guide on how to set up authentication for cloud providers,
+see the integration [docs](/flux/integrations/).
+
 ### Prefix
 
 `.spec.prefix` is an optional field to enable server-side filtering

internal/controller/bucket_controller.go

@@ -44,6 +44,8 @@ import (
 
 	eventv1 "github.com/fluxcd/pkg/apis/event/v1beta1"
 	"github.com/fluxcd/pkg/apis/meta"
+	"github.com/fluxcd/pkg/auth"
+	"github.com/fluxcd/pkg/cache"
 	"github.com/fluxcd/pkg/runtime/conditions"
 	helper "github.com/fluxcd/pkg/runtime/controller"
 	"github.com/fluxcd/pkg/runtime/jitter"
@@ -116,6 +118,8 @@ var bucketFailConditions = []string{
 // +kubebuilder:rbac:groups=source.toolkit.fluxcd.io,resources=buckets/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=source.toolkit.fluxcd.io,resources=buckets/finalizers,verbs=get;create;update;patch;delete
 // +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch
+// +kubebuilder:rbac:groups="",resources=serviceaccounts,verbs=get;list;watch
+// +kubebuilder:rbac:groups="",resources=serviceaccounts/token,verbs=create
 
 // BucketReconciler reconciles a v1.Bucket object.
 type BucketReconciler struct {
@@ -125,6 +129,7 @@ type BucketReconciler struct {
 
 	Storage        *Storage
 	ControllerName string
+	TokenCache     *cache.TokenCache
 
 	patchOptions []patch.Option
 }
@@ -430,6 +435,18 @@ func (r *BucketReconciler) reconcileStorage(ctx context.Context, sp *patch.Seria
 // the provider. If this fails, it records v1.FetchFailedCondition=True on
 // the object and returns early.
 func (r *BucketReconciler) reconcileSource(ctx context.Context, sp *patch.SerialPatcher, obj *sourcev1.Bucket, index *index.Digester, dir string) (sreconcile.Result, error) {
+	usesObjectLevelWorkloadIdentity := obj.Spec.Provider != "" && obj.Spec.Provider != sourcev1.BucketProviderGeneric && obj.Spec.ServiceAccountName != ""
+	if usesObjectLevelWorkloadIdentity {
+		if !auth.IsObjectLevelWorkloadIdentityEnabled() {
+			const gate = auth.FeatureGateObjectLevelWorkloadIdentity
+			const msgFmt = "to use spec.serviceAccountName for provider authentication please enable the %s feature gate in the controller"
+			err := fmt.Errorf(msgFmt, gate)
+			e := serror.NewStalling(err, meta.FeatureGateDisabledReason)
+			conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, e.Reason, "%s", e)
+			return sreconcile.ResultEmpty, e
+		}
+	}
+
 	creds, err := r.setupCredentials(ctx, obj)
 	if err != nil {
 		e := serror.NewGeneric(err, sourcev1.AuthenticationFailedReason)
@@ -590,6 +607,10 @@ func (r *BucketReconciler) reconcileDelete(ctx context.Context, obj *sourcev1.Bu
 	// Remove our finalizer from the list
 	controllerutil.RemoveFinalizer(obj, sourcev1.SourceFinalizer)
 
+	// Cleanup caches.
+	r.TokenCache.DeleteEventsForObject(sourcev1.BucketKind,
+		obj.GetName(), obj.GetNamespace(), cache.OperationReconcile)
+
 	// Stop reconciliation as the object is being deleted
 	return sreconcile.ResultEmpty, nil
 }
@@ -838,19 +859,47 @@ func (r *BucketReconciler) setupCredentials(ctx context.Context, obj *sourcev1.B
 // createBucketProvider creates a provider-specific bucket client using the given credentials and configuration.
 // It handles different bucket providers (AWS, GCP, Azure, generic) and returns the appropriate client.
 func (r *BucketReconciler) createBucketProvider(ctx context.Context, obj *sourcev1.Bucket, creds *bucketCredentials) (BucketProvider, error) {
+	var authOpts []auth.Option
+
+	if obj.Spec.ServiceAccountName != "" {
+		serviceAccount := client.ObjectKey{
+			Name:      obj.Spec.ServiceAccountName,
+			Namespace: obj.GetNamespace(),
+		}
+		authOpts = append(authOpts, auth.WithServiceAccount(serviceAccount, r.Client))
+	}
+
+	if r.TokenCache != nil {
+		involvedObject := cache.InvolvedObject{
+			Kind:      sourcev1.BucketKind,
+			Name:      obj.GetName(),
+			Namespace: obj.GetNamespace(),
+			Operation: cache.OperationReconcile,
+		}
+		authOpts = append(authOpts, auth.WithCache(*r.TokenCache, involvedObject))
+	}
+
+	if creds.proxyURL != nil {
+		authOpts = append(authOpts, auth.WithProxyURL(*creds.proxyURL))
+	}
+
 	switch obj.Spec.Provider {
 	case sourcev1.BucketProviderGoogle:
-		if err := gcp.ValidateSecret(creds.secret); err != nil {
-			return nil, err
-		}
 		var opts []gcp.Option
-		if creds.secret != nil {
-			opts = append(opts, gcp.WithSecret(creds.secret))
-		}
 		if creds.proxyURL != nil {
 			opts = append(opts, gcp.WithProxyURL(creds.proxyURL))
 		}
-		return gcp.NewClient(ctx, opts...)
+
+		if creds.secret != nil {
+			if err := gcp.ValidateSecret(creds.secret); err != nil {
+				return nil, err
+			}
+			opts = append(opts, gcp.WithSecret(creds.secret))
+		} else {
+			opts = append(opts, gcp.WithAuth(authOpts...))
+		}
+
+		return gcp.NewClient(ctx, obj, opts...)
 
 	case sourcev1.BucketProviderAzure:
 		if err := azure.ValidateSecret(creds.secret); err != nil {

internal/controller/bucket_controller_test.go

@@ -437,6 +437,7 @@ func TestBucketReconciler_reconcileSource_generic(t *testing.T) {
 		bucketObjects    []*s3mock.Object
 		middleware       http.Handler
 		secret           *corev1.Secret
+		serviceAccount   *corev1.ServiceAccount
 		beforeFunc       func(obj *sourcev1.Bucket)
 		want             sreconcile.Result
 		wantErr          bool
@@ -910,6 +911,10 @@ func TestBucketReconciler_reconcileSource_generic(t *testing.T) {
 				clientBuilder.WithObjects(tt.secret)
 			}
 
+			if tt.serviceAccount != nil {
+				clientBuilder.WithObjects(tt.serviceAccount)
+			}
+
 			r := &BucketReconciler{
 				EventRecorder: record.NewFakeRecorder(32),
 				Client:        clientBuilder.Build(),
@@ -972,15 +977,17 @@ func TestBucketReconciler_reconcileSource_generic(t *testing.T) {
 
 func TestBucketReconciler_reconcileSource_gcs(t *testing.T) {
 	tests := []struct {
-		name             string
-		bucketName       string
-		bucketObjects    []*gcsmock.Object
-		secret           *corev1.Secret
-		beforeFunc       func(obj *sourcev1.Bucket)
-		want             sreconcile.Result
-		wantErr          bool
-		assertIndex      *index.Digester
-		assertConditions []metav1.Condition
+		name                               string
+		bucketName                         string
+		bucketObjects                      []*gcsmock.Object
+		secret                             *corev1.Secret
+		serviceAccount                     *corev1.ServiceAccount
+		beforeFunc                         func(obj *sourcev1.Bucket)
+		want                               sreconcile.Result
+		wantErr                            bool
+		assertIndex                        *index.Digester
+		assertConditions                   []metav1.Condition
+		disableObjectLevelWorkloadIdentity bool
 	}{
 		{
 			name:       "Reconciles GCS source",
@@ -1283,6 +1290,80 @@ func TestBucketReconciler_reconcileSource_gcs(t *testing.T) {
 				*conditions.UnknownCondition(meta.ReadyCondition, meta.ProgressingReason, "building artifact: new upstream revision 'sha256:b4c2a60ce44b67f5b659a95ce4e4cc9e2a86baf13afb72bd397c5384cbc0e479'"),
 			},
 		},
+		{
+			name:       "GCS Object-Level Workload Identity (no secret)",
+			bucketName: "dummy",
+			bucketObjects: []*gcsmock.Object{
+				{
+					Key:         "test.txt",
+					ContentType: "text/plain",
+					Content:     []byte("test"),
+					Generation:  3,
+				},
+			},
+			serviceAccount: &corev1.ServiceAccount{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-sa",
+				},
+			},
+			beforeFunc: func(obj *sourcev1.Bucket) {
+				obj.Spec.ServiceAccountName = "test-sa"
+			},
+			want: sreconcile.ResultSuccess,
+			assertIndex: index.NewDigester(index.WithIndex(map[string]string{
+				"test.txt": "098f6bcd4621d373cade4e832627b4f6",
+			})),
+			assertConditions: []metav1.Condition{
+				*conditions.TrueCondition(meta.ReconcilingCondition, meta.ProgressingReason, "building artifact: new upstream revision 'sha256:b4c2a60ce44b67f5b659a95ce4e4cc9e2a86baf13afb72bd397c5384cbc0e479'"),
+				*conditions.UnknownCondition(meta.ReadyCondition, meta.ProgressingReason, "building artifact: new upstream revision 'sha256:b4c2a60ce44b67f5b659a95ce4e4cc9e2a86baf13afb72bd397c5384cbc0e479'"),
+			},
+		},
+		{
+			name:       "GCS Controller-Level Workload Identity (no secret, no SA)",
+			bucketName: "dummy",
+			bucketObjects: []*gcsmock.Object{
+				{
+					Key:         "test.txt",
+					ContentType: "text/plain",
+					Content:     []byte("test"),
+					Generation:  3,
+				},
+			},
+			beforeFunc: func(obj *sourcev1.Bucket) {
+				// ServiceAccountName は設定しない (Controller-Level)
+			},
+			want: sreconcile.ResultSuccess,
+			assertIndex: index.NewDigester(index.WithIndex(map[string]string{
+				"test.txt": "098f6bcd4621d373cade4e832627b4f6",
+			})),
+			assertConditions: []metav1.Condition{
+				*conditions.TrueCondition(meta.ReconcilingCondition, meta.ProgressingReason, "building artifact: new upstream revision 'sha256:b4c2a60ce44b67f5b659a95ce4e4cc9e2a86baf13afb72bd397c5384cbc0e479'"),
+				*conditions.UnknownCondition(meta.ReadyCondition, meta.ProgressingReason, "building artifact: new upstream revision 'sha256:b4c2a60ce44b67f5b659a95ce4e4cc9e2a86baf13afb72bd397c5384cbc0e479'"),
+			},
+		},
+		{
+			name:       "GCS Object-Level fails when feature gate disabled",
+			bucketName: "dummy",
+			serviceAccount: &corev1.ServiceAccount{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-sa",
+				},
+			},
+			beforeFunc: func(obj *sourcev1.Bucket) {
+				obj.Spec.ServiceAccountName = "test-sa"
+				conditions.MarkReconciling(obj, meta.ProgressingReason, "foo")
+				conditions.MarkUnknown(obj, meta.ReadyCondition, "foo", "bar")
+			},
+			want:        sreconcile.ResultEmpty,
+			wantErr:     true,
+			assertIndex: index.NewDigester(),
+			assertConditions: []metav1.Condition{
+				*conditions.TrueCondition(sourcev1.FetchFailedCondition, meta.FeatureGateDisabledReason, "to use spec.serviceAccountName for provider authentication please enable the ObjectLevelWorkloadIdentity feature gate in the controller"),
+				*conditions.TrueCondition(meta.ReconcilingCondition, meta.ProgressingReason, "foo"),
+				*conditions.UnknownCondition(meta.ReadyCondition, "foo", "bar"),
+			},
+			disableObjectLevelWorkloadIdentity: true,
+		},
 		// TODO: Middleware for mock server to test authentication using secret.
 	}
 	for _, tt := range tests {
@@ -1297,12 +1378,24 @@ func TestBucketReconciler_reconcileSource_gcs(t *testing.T) {
 				clientBuilder.WithObjects(tt.secret)
 			}
 
+			if tt.serviceAccount != nil {
+				clientBuilder.WithObjects(tt.serviceAccount)
+			}
+
 			r := &BucketReconciler{
 				EventRecorder: record.NewFakeRecorder(32),
 				Client:        clientBuilder.Build(),
 				Storage:       testStorage,
 				patchOptions:  getPatchOptions(bucketReadyCondition.Owned, "sc"),
 			}
+
+			// Handle ObjectLevelWorkloadIdentity feature gate environment variable
+			if tt.disableObjectLevelWorkloadIdentity {
+				t.Setenv("ENABLE_OBJECT_LEVEL_WORKLOAD_IDENTITY", "false")
+			} else if tt.serviceAccount != nil {
+				t.Setenv("ENABLE_OBJECT_LEVEL_WORKLOAD_IDENTITY", "true")
+			}
+
 			tmpDir := t.TempDir()
 
 			// Test bucket object.

main.go

@@ -272,6 +272,7 @@ func main() {
 		Metrics:        metrics,
 		Storage:        storage,
 		ControllerName: controllerName,
+		TokenCache:     tokenCache,
 	}).SetupWithManagerAndOptions(mgr, controller.BucketReconcilerOptions{
 		RateLimiter: helper.GetRateLimiter(rateLimiterOptions),
 	}); err != nil {

pkg/gcp/gcp.go

@@ -34,6 +34,11 @@ import (
 	htransport "google.golang.org/api/transport/http"
 	corev1 "k8s.io/api/core/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
+
+	"github.com/fluxcd/pkg/auth"
+	gcpauth "github.com/fluxcd/pkg/auth/gcp"
+
+	sourcev1 "github.com/fluxcd/source-controller/api/v1"
 )
 
 var (
@@ -69,13 +74,21 @@ func WithProxyURL(proxyURL *url.URL) Option {
 	}
 }
 
+// WithAuth sets the auth options for workload identity authentication.
+func WithAuth(authOpts ...auth.Option) Option {
+	return func(o *options) {
+		o.authOpts = authOpts
+	}
+}
+
 type options struct {
 	secret   *corev1.Secret
 	proxyURL *url.URL
+	authOpts []auth.Option
 
 	// newCustomHTTPClient should create a new HTTP client for interacting with the GCS API.
 	// This is a test-only option required for mocking the real logic, which requires either
-	// a valid Google Service Account Key or ADC. Both are not available in tests.
+	// a valid Google Service Account Key or Controller-Level Workload Identity. Both are not available in tests.
 	// The real logic is implemented in the newHTTPClient function, which is used when
 	// constructing the default options object.
 	newCustomHTTPClient func(context.Context, *options) (*http.Client, error)
@@ -89,7 +102,7 @@ func newOptions() *options {
 
 // NewClient creates a new GCP storage client. The Client will automatically look for the Google Application
 // Credential environment variable or look for the Google Application Credential file.
-func NewClient(ctx context.Context, opts ...Option) (*GCSClient, error) {
+func NewClient(ctx context.Context, bucket *sourcev1.Bucket, opts ...Option) (*GCSClient, error) {
 	o := newOptions()
 	for _, opt := range opts {
 		opt(o)
@@ -100,7 +113,10 @@ func NewClient(ctx context.Context, opts ...Option) (*GCSClient, error) {
 	switch {
 	case o.secret != nil && o.proxyURL == nil:
 		clientOpts = append(clientOpts, option.WithCredentialsJSON(o.secret.Data["serviceaccount"]))
-	case o.proxyURL != nil:
+	case o.secret == nil && o.proxyURL == nil:
+		tokenSource := gcpauth.NewTokenSource(ctx, o.authOpts...)
+		clientOpts = append(clientOpts, option.WithTokenSource(tokenSource))
+	default: // o.proxyURL != nil:
 		httpClient, err := o.newCustomHTTPClient(ctx, o)
 		if err != nil {
 			return nil, err
@@ -135,6 +151,9 @@ func newHTTPClient(ctx context.Context, o *options) (*http.Client, error) {
 			return nil, fmt.Errorf("failed to create Google credentials from secret: %w", err)
 		}
 		opts = append(opts, option.WithCredentials(creds))
+	} else { // Workload Identity.
+		tokenSource := gcpauth.NewTokenSource(ctx, o.authOpts...)
+		opts = append(opts, option.WithTokenSource(tokenSource))
 	}
 
 	transport, err := htransport.NewTransport(ctx, baseTransport, opts...)

pkg/gcp/gcp_test.go

@@ -42,6 +42,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	sourcev1 "github.com/fluxcd/source-controller/api/v1"
 	testproxy "github.com/fluxcd/source-controller/tests/proxy"
 )
 
@@ -82,6 +83,22 @@ var (
 	}
 )
 
+// createTestBucket creates a test bucket for testing purposes
+func createTestBucket() *sourcev1.Bucket {
+	return &sourcev1.Bucket{
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "test-bucket",
+			Namespace: "default",
+		},
+		Spec: sourcev1.BucketSpec{
+			BucketName: bucketName,
+			Endpoint:   "storage.googleapis.com",
+			Provider:   sourcev1.BucketProviderGoogle,
+			Interval:   v1.Duration{Duration: time.Minute * 5},
+		},
+	}
+}
+
 func TestMain(m *testing.M) {
 	hc, host, close = newTestServer(func(w http.ResponseWriter, r *http.Request) {
 		io.Copy(io.Discard, r.Body)
@@ -147,7 +164,8 @@ func TestMain(m *testing.M) {
 }
 
 func TestNewClientWithSecretErr(t *testing.T) {
-	gcpClient, err := NewClient(context.Background(), WithSecret(secret.DeepCopy()))
+	bucket := createTestBucket()
+	gcpClient, err := NewClient(context.Background(), bucket, WithSecret(secret.DeepCopy()))
 	t.Log(err)
 	assert.Error(t, err, "dialing: invalid character 'e' looking for beginning of value")
 	assert.Assert(t, gcpClient == nil)
@@ -158,31 +176,29 @@ func TestNewClientWithProxyErr(t *testing.T) {
 	assert.Assert(t, !envADCIsSet)
 	assert.Assert(t, !metadata.OnGCE())
 
-	tests := []struct {
-		name string
-		opts []Option
-		err  string
-	}{
-		{
-			name: "invalid secret",
-			opts: []Option{WithSecret(secret.DeepCopy())},
-			err:  "failed to create Google credentials from secret: invalid character 'e' looking for beginning of value",
-		},
-		{
-			name: "attempts default credentials",
-			err:  "failed to create Google HTTP transport: google: could not find default credentials. See https://cloud.google.com/docs/authentication/external/set-up-adc for more information",
-		},
-	}
+	t.Run("with secret", func(t *testing.T) {
+		g := NewWithT(t)
+		bucket := createTestBucket()
+		gcpClient, err := NewClient(context.Background(), bucket,
+			WithProxyURL(&url.URL{}),
+			WithSecret(secret.DeepCopy()))
+		g.Expect(err).To(HaveOccurred())
+		g.Expect(gcpClient).To(BeNil())
+		g.Expect(err.Error()).To(Equal("failed to create Google credentials from secret: invalid character 'e' looking for beginning of value"))
+	})
 
-	for _, tt := range tests {
-		tt := tt
-		t.Run(tt.name, func(t *testing.T) {
-			opts := append([]Option{WithProxyURL(&url.URL{})}, tt.opts...)
-			gcpClient, err := NewClient(context.Background(), opts...)
-			assert.Error(t, err, tt.err)
-			assert.Assert(t, gcpClient == nil)
-		})
-	}
+	t.Run("without secret", func(t *testing.T) {
+		g := NewWithT(t)
+		bucket := createTestBucket()
+		gcpClient, err := NewClient(context.Background(), bucket,
+			WithProxyURL(&url.URL{}))
+		g.Expect(err).NotTo(HaveOccurred())
+		g.Expect(gcpClient).NotTo(BeNil())
+		bucketAttrs, err := gcpClient.Client.Bucket("some-bucket").Attrs(context.Background())
+		g.Expect(err).To(HaveOccurred())
+		g.Expect(bucketAttrs).To(BeNil())
+		g.Expect(err.Error()).To(ContainSubstring("failed to create provider access token"))
+	})
 }
 
 func TestProxy(t *testing.T) {
@@ -224,7 +240,8 @@ func TestProxy(t *testing.T) {
 					return &http.Client{Transport: transport}, nil
 				}
 			})
-			gcpClient, err := NewClient(context.Background(), opts...)
+			bucket := createTestBucket()
+			gcpClient, err := NewClient(context.Background(), bucket, opts...)
 			assert.NilError(t, err)
 			assert.Assert(t, gcpClient != nil)
 			gcpClient.Client.SetRetry(gcpstorage.WithMaxAttempts(1))