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

Signed-off-by: cappyzawa <[email protected]>

Author: cappyzawa

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,37 +647,55 @@ 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.
+##### GCP Controller-Level Workload Identity example
 
-The Provider allows for specifying the
-[Bucket location](https://cloud.google.com/storage/docs/locations) using the
-[`.spec.region` field](#region).
+```yaml
+---
+apiVersion: source.toolkit.fluxcd.io/v1
+kind: Bucket
+metadata:
+  name: gcp-controller-level-workload-identity
+  namespace: default
+spec:
+  interval: 5m0s
+  provider: gcp
+  bucketName: podinfo
+  endpoint: storage.googleapis.com
+  region: us-east-1
+  timeout: 30s
+```
+
+##### GCP Object-Level Workload Identity example
 
-##### GCP 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-workload-identity
+  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
@@ -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 {