Add proxy support for Azure buckets

api/v1beta2/bucket_types.go

@@ -113,7 +113,7 @@ type BucketSpec struct {
 	// ProxySecretRef specifies the Secret containing the proxy configuration
 	// to use while communicating with the Bucket server.
 	//
-	// Only supported for the `generic` and `gcp` providers.
+	// Only supported for the `generic`, `gcp` and `azure` providers.
 	// +optional
 	ProxySecretRef *meta.LocalObjectReference `json:"proxySecretRef,omitempty"`
 

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

@@ -397,7 +397,7 @@ spec:
                   to use while communicating with the Bucket server.
 
 
-                  Only supported for the `generic` and `gcp` providers.
+                  Only supported for the `generic`, `gcp` and `azure` providers.
                 properties:
                   name:
                     description: Name of the referent.

docs/api/v1beta2/source.md

@@ -219,7 +219,7 @@ github.com/fluxcd/pkg/apis/meta.LocalObjectReference
 <em>(Optional)</em>
 <p>ProxySecretRef specifies the Secret containing the proxy configuration
 to use while communicating with the Bucket server.</p>
-<p>Only supported for the <code>generic</code> and <code>gcp</code> providers.</p>
+<p>Only supported for the <code>generic</code>, <code>gcp</code> and <code>azure</code> providers.</p>
 </td>
 </tr>
 <tr>
@@ -1648,7 +1648,7 @@ github.com/fluxcd/pkg/apis/meta.LocalObjectReference
 <em>(Optional)</em>
 <p>ProxySecretRef specifies the Secret containing the proxy configuration
 to use while communicating with the Bucket server.</p>
-<p>Only supported for the <code>generic</code> and <code>gcp</code> providers.</p>
+<p>Only supported for the <code>generic</code>, <code>gcp</code> and <code>azure</code> providers.</p>
 </td>
 </tr>
 <tr>

docs/spec/v1beta2/buckets.md

@@ -854,7 +854,7 @@ The Secret can contain three keys:
 - `password`, to specify the password to use if the proxy server is protected by
    basic authentication. This is an optional key.
 
-This API is only supported for the `generic` and `gcp` [providers](#provider).
+This API is only supported for the `generic`, `gcp` and `azure` [providers](#provider).
 
 Example:
 

internal/controller/bucket_controller.go

@@ -465,7 +465,14 @@ func (r *BucketReconciler) reconcileSource(ctx context.Context, sp *patch.Serial
 			conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, e.Reason, "%s", e)
 			return sreconcile.ResultEmpty, e
 		}
-		if provider, err = azure.NewClient(obj, secret); err != nil {
+		var opts []azure.Option
+		if secret != nil {
+			opts = append(opts, azure.WithSecret(secret))
+		}
+		if proxyURL != nil {
+			opts = append(opts, azure.WithProxyURL(proxyURL))
+		}
+		if provider, err = azure.NewClient(obj, opts...); err != nil {
 			e := serror.NewGeneric(err, "ClientError")
 			conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, e.Reason, "%s", e)
 			return sreconcile.ResultEmpty, e

pkg/azure/blob.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
 	"net/url"
 	"os"
 	"path/filepath"
@@ -64,6 +65,48 @@ type BlobClient struct {
 	*azblob.Client
 }
 
+// Option configures the BlobClient.
+type Option func(*options)
+
+// WithSecret sets the Secret to use for the BlobClient.
+func WithSecret(secret *corev1.Secret) Option {
+	return func(o *options) {
+		o.secret = secret
+	}
+}
+
+// WithProxyURL sets the proxy URL to use for the BlobClient.
+func WithProxyURL(proxyURL *url.URL) Option {
+	return func(o *options) {
+		o.proxyURL = proxyURL
+	}
+}
+
+type options struct {
+	secret             *corev1.Secret
+	proxyURL           *url.URL
+	withoutCredentials bool
+	withoutRetries     bool
+}
+
+// withoutCredentials forces the BlobClient to not use any credentials.
+// This is a test-only option useful for testing the client with HTTP
+// endpoints (without TLS) alongside all the other options unrelated to
+// credentials.
+func withoutCredentials() Option {
+	return func(o *options) {
+		o.withoutCredentials = true
+	}
+}
+
+// withoutRetries sets the BlobClient to not retry requests.
+// This is a test-only option useful for testing connection errors.
+func withoutRetries() Option {
+	return func(o *options) {
+		o.withoutRetries = true
+	}
+}
+
 // NewClient creates a new Azure Blob storage client.
 // The credential config on the client is set based on the data from the
 // Bucket and Secret. It detects credentials in the Secret in the following
@@ -87,56 +130,80 @@ type BlobClient struct {
 //
 // If no credentials are found, and the azidentity.ChainedTokenCredential can
 // not be established. A simple client without credentials is returned.
-func NewClient(obj *sourcev1.Bucket, secret *corev1.Secret) (c *BlobClient, err error) {
+func NewClient(obj *sourcev1.Bucket, opts ...Option) (c *BlobClient, err error) {
 	c = &BlobClient{}
 
+	var o options
+	for _, opt := range opts {
+		opt(&o)
+	}
+
+	clientOpts := &azblob.ClientOptions{}
+
+	if o.proxyURL != nil {
+		transport := http.DefaultTransport.(*http.Transport).Clone()
+		transport.Proxy = http.ProxyURL(o.proxyURL)
+		clientOpts.ClientOptions.Transport = &http.Client{Transport: transport}
+	}
+
+	if o.withoutRetries {
+		clientOpts.ClientOptions.Retry.ShouldRetry = func(resp *http.Response, err error) bool {
+			return false
+		}
+	}
+
+	if o.withoutCredentials {
+		c.Client, err = azblob.NewClientWithNoCredential(obj.Spec.Endpoint, clientOpts)
+		return
+	}
+
 	var token azcore.TokenCredential
 
-	if secret != nil && len(secret.Data) > 0 {
+	if o.secret != nil && len(o.secret.Data) > 0 {
 		// Attempt AAD Token Credential options first.
-		if token, err = tokenCredentialFromSecret(secret); err != nil {
-			err = fmt.Errorf("failed to create token credential from '%s' Secret: %w", secret.Name, err)
+		if token, err = tokenCredentialFromSecret(o.secret); err != nil {
+			err = fmt.Errorf("failed to create token credential from '%s' Secret: %w", o.secret.Name, err)
 			return
 		}
 		if token != nil {
-			c.Client, err = azblob.NewClient(obj.Spec.Endpoint, token, nil)
+			c.Client, err = azblob.NewClient(obj.Spec.Endpoint, token, clientOpts)
 			return
 		}
 
 		// Fallback to Shared Key Credential.
 		var cred *azblob.SharedKeyCredential
-		if cred, err = sharedCredentialFromSecret(obj.Spec.Endpoint, secret); err != nil {
+		if cred, err = sharedCredentialFromSecret(obj.Spec.Endpoint, o.secret); err != nil {
 			return
 		}
 		if cred != nil {
-			c.Client, err = azblob.NewClientWithSharedKeyCredential(obj.Spec.Endpoint, cred, &azblob.ClientOptions{})
+			c.Client, err = azblob.NewClientWithSharedKeyCredential(obj.Spec.Endpoint, cred, clientOpts)
 			return
 		}
 
 		var fullPath string
-		if fullPath, err = sasTokenFromSecret(obj.Spec.Endpoint, secret); err != nil {
+		if fullPath, err = sasTokenFromSecret(obj.Spec.Endpoint, o.secret); err != nil {
 			return
 		}
 
-		c.Client, err = azblob.NewClientWithNoCredential(fullPath, &azblob.ClientOptions{})
+		c.Client, err = azblob.NewClientWithNoCredential(fullPath, clientOpts)
 		return
 	}
 
 	// Compose token chain based on environment.
 	// This functions as a replacement for azidentity.NewDefaultAzureCredential
 	// to not shell out.
-	token, err = chainCredentialWithSecret(secret)
+	token, err = chainCredentialWithSecret(o.secret)
 	if err != nil {
 		err = fmt.Errorf("failed to create environment credential chain: %w", err)
 		return nil, err
 	}
 	if token != nil {
-		c.Client, err = azblob.NewClient(obj.Spec.Endpoint, token, nil)
+		c.Client, err = azblob.NewClient(obj.Spec.Endpoint, token, clientOpts)
 		return
 	}
 
 	// Fallback to simple client.
-	c.Client, err = azblob.NewClientWithNoCredential(obj.Spec.Endpoint, nil)
+	c.Client, err = azblob.NewClientWithNoCredential(obj.Spec.Endpoint, clientOpts)
 	return
 }
 

pkg/azure/blob_integration_test.go

@@ -94,7 +94,7 @@ func TestMain(m *testing.M) {
 func TestBlobClient_BucketExists(t *testing.T) {
 	g := NewWithT(t)
 
-	client, err := NewClient(testBucket.DeepCopy(), testSecret.DeepCopy())
+	client, err := NewClient(testBucket.DeepCopy(), WithSecret(testSecret.DeepCopy()))
 	g.Expect(err).ToNot(HaveOccurred())
 	g.Expect(client).ToNot(BeNil())
 
@@ -120,7 +120,7 @@ func TestBlobClient_BucketExists(t *testing.T) {
 func TestBlobClient_BucketNotExists(t *testing.T) {
 	g := NewWithT(t)
 
-	client, err := NewClient(testBucket.DeepCopy(), testSecret.DeepCopy())
+	client, err := NewClient(testBucket.DeepCopy(), WithSecret(testSecret.DeepCopy()))
 	g.Expect(err).ToNot(HaveOccurred())
 	g.Expect(client).ToNot(BeNil())
 
@@ -140,7 +140,7 @@ func TestBlobClient_FGetObject(t *testing.T) {
 
 	tempDir := t.TempDir()
 
-	client, err := NewClient(testBucket.DeepCopy(), testSecret.DeepCopy())
+	client, err := NewClient(testBucket.DeepCopy(), WithSecret(testSecret.DeepCopy()))
 	g.Expect(err).ToNot(HaveOccurred())
 	g.Expect(client).ToNot(BeNil())
 
@@ -180,7 +180,7 @@ func TestBlobClientSASKey_FGetObject(t *testing.T) {
 	tempDir := t.TempDir()
 
 	// create a client with the shared key
-	client, err := NewClient(testBucket.DeepCopy(), testSecret.DeepCopy())
+	client, err := NewClient(testBucket.DeepCopy(), WithSecret(testSecret.DeepCopy()))
 	g.Expect(err).ToNot(HaveOccurred())
 	g.Expect(client).ToNot(BeNil())
 
@@ -221,7 +221,7 @@ func TestBlobClientSASKey_FGetObject(t *testing.T) {
 		},
 	}
 
-	sasKeyClient, err := NewClient(testBucket.DeepCopy(), testSASKeySecret.DeepCopy())
+	sasKeyClient, err := NewClient(testBucket.DeepCopy(), WithSecret(testSASKeySecret.DeepCopy()))
 	g.Expect(err).ToNot(HaveOccurred())
 
 	// Test if bucket and blob exists using sasKey.
@@ -246,7 +246,7 @@ func TestBlobClientContainerSASKey_BucketExists(t *testing.T) {
 	g := NewWithT(t)
 
 	// create a client with the shared key
-	client, err := NewClient(testBucket.DeepCopy(), testSecret.DeepCopy())
+	client, err := NewClient(testBucket.DeepCopy(), WithSecret(testSecret.DeepCopy()))
 	g.Expect(err).ToNot(HaveOccurred())
 	g.Expect(client).ToNot(BeNil())
 
@@ -286,7 +286,7 @@ func TestBlobClientContainerSASKey_BucketExists(t *testing.T) {
 		},
 	}
 
-	sasKeyClient, err := NewClient(testBucket.DeepCopy(), testSASKeySecret.DeepCopy())
+	sasKeyClient, err := NewClient(testBucket.DeepCopy(), WithSecret(testSASKeySecret.DeepCopy()))
 	g.Expect(err).ToNot(HaveOccurred())
 
 	ctx, timeout = context.WithTimeout(context.Background(), testTimeout)
@@ -308,7 +308,7 @@ func TestBlobClientContainerSASKey_BucketExists(t *testing.T) {
 func TestBlobClient_FGetObject_NotFoundErr(t *testing.T) {
 	g := NewWithT(t)
 
-	client, err := NewClient(testBucket.DeepCopy(), testSecret.DeepCopy())
+	client, err := NewClient(testBucket.DeepCopy(), WithSecret(testSecret.DeepCopy()))
 	g.Expect(err).ToNot(HaveOccurred())
 	g.Expect(client).ToNot(BeNil())
 
@@ -335,7 +335,7 @@ func TestBlobClient_FGetObject_NotFoundErr(t *testing.T) {
 func TestBlobClient_VisitObjects(t *testing.T) {
 	g := NewWithT(t)
 
-	client, err := NewClient(testBucket.DeepCopy(), testSecret.DeepCopy())
+	client, err := NewClient(testBucket.DeepCopy(), WithSecret(testSecret.DeepCopy()))
 	g.Expect(err).ToNot(HaveOccurred())
 	g.Expect(client).ToNot(BeNil())
 
@@ -375,7 +375,7 @@ func TestBlobClient_VisitObjects(t *testing.T) {
 func TestBlobClient_VisitObjects_CallbackErr(t *testing.T) {
 	g := NewWithT(t)
 
-	client, err := NewClient(testBucket.DeepCopy(), testSecret.DeepCopy())
+	client, err := NewClient(testBucket.DeepCopy(), WithSecret(testSecret.DeepCopy()))
 	g.Expect(err).ToNot(HaveOccurred())
 	g.Expect(client).ToNot(BeNil())