[release/v1.9.x] fix: harden Bucket reconciliation error paths

internal/bucket/azure/blob.go

@@ -335,7 +335,13 @@ func (c *BlobClient) FGetObject(ctx context.Context, bucketName, objectName, loc
 		return "", err
 	}
 
-	return string(*res.ETag), nil
+	// The ETag response header is optional and may be absent from
+	// non-conformant storage endpoints.
+	var etag string
+	if res.ETag != nil {
+		etag = string(*res.ETag)
+	}
+	return etag, nil
 }
 
 // VisitObjects iterates over the items in the provided object storage
@@ -354,8 +360,20 @@ func (c *BlobClient) VisitObjects(ctx context.Context, bucketName string, prefix
 			err = fmt.Errorf("listing objects from bucket '%s' failed: %w", bucketName, err)
 			return err
 		}
+		if resp.Segment == nil {
+			continue
+		}
 		for _, blob := range resp.Segment.BlobItems {
-			if err := visit(*blob.Name, fmt.Sprintf("%x", *blob.Properties.ETag)); err != nil {
+			// The list response fields are optional and may be absent
+			// from non-conformant storage endpoints.
+			if blob == nil || blob.Name == nil {
+				continue
+			}
+			var etag string
+			if blob.Properties != nil && blob.Properties.ETag != nil {
+				etag = fmt.Sprintf("%x", *blob.Properties.ETag)
+			}
+			if err := visit(*blob.Name, etag); err != nil {
 				err = fmt.Errorf("listing objects from bucket '%s' failed: %w", bucketName, err)
 				return err
 			}

internal/bucket/azure/blob_test.go

@@ -28,6 +28,8 @@ import (
 	"math/big"
 	"net/http"
 	"net/url"
+	"os"
+	"path/filepath"
 	"testing"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
@@ -551,6 +553,109 @@ func TestBlobClient_VisitObjects_Prefix(t *testing.T) {
 	}
 }
 
+func TestBlobClient_FGetObject_MissingEtag(t *testing.T) {
+	g := NewWithT(t)
+
+	bucketName := "test-bucket"
+	objectName := "file.txt"
+
+	// start mock bucket server which omits the ETag response header
+	bucketListener, bucketAddr, _ := testlistener.New(t)
+	bucketEndpoint := fmt.Sprintf("http://%s", bucketAddr)
+	bucketHandler := http.NewServeMux()
+	bucketHandler.HandleFunc(fmt.Sprintf("GET /%s/%s", bucketName, objectName), func(w http.ResponseWriter, r *http.Request) {
+		_, err := w.Write([]byte("file contents"))
+		g.Expect(err).ToNot(HaveOccurred())
+	})
+	bucketServer := &http.Server{
+		Addr:    bucketAddr,
+		Handler: bucketHandler,
+	}
+	go bucketServer.Serve(bucketListener)
+	defer bucketServer.Shutdown(context.Background())
+
+	bucket := &sourcev1.Bucket{
+		Spec: sourcev1.BucketSpec{
+			Endpoint: bucketEndpoint,
+		},
+	}
+	client, err := NewClient(t.Context(),
+		bucket,
+		withoutCredentials(),
+		withoutRetries())
+	g.Expect(err).ToNot(HaveOccurred())
+
+	localPath := filepath.Join(t.TempDir(), objectName)
+	etag, err := client.FGetObject(t.Context(), bucketName, objectName, localPath)
+	g.Expect(err).ToNot(HaveOccurred())
+	g.Expect(etag).To(BeEmpty())
+	g.Expect(os.ReadFile(localPath)).To(BeEquivalentTo("file contents"))
+}
+
+func TestBlobClient_VisitObjects_MissingFields(t *testing.T) {
+	g := NewWithT(t)
+
+	bucketName := "test-bucket"
+
+	// start mock bucket server whose listing contains entries with
+	// missing optional fields
+	bucketListener, bucketAddr, _ := testlistener.New(t)
+	bucketEndpoint := fmt.Sprintf("http://%s", bucketAddr)
+	bucketHandler := http.NewServeMux()
+	bucketHandler.HandleFunc(fmt.Sprintf("GET /%s", bucketName), func(w http.ResponseWriter, r *http.Request) {
+		resp := fmt.Sprintf(`<?xml version="1.0" encoding="utf-8"?>
+<EnumerationResults ContainerName="%s/%s">
+<Blobs>
+  <Blob>
+    <Name>no-properties.txt</Name>
+  </Blob>
+  <Blob>
+    <Name>no-etag.txt</Name>
+    <Properties />
+  </Blob>
+  <Blob>
+    <Properties>
+      <Etag>0x8D9B2A2A2A2A2A2</Etag>
+    </Properties>
+  </Blob>
+</Blobs>
+<NextMarker />
+</EnumerationResults>`, bucketEndpoint, bucketName)
+		_, err := w.Write([]byte(resp))
+		g.Expect(err).ToNot(HaveOccurred())
+	})
+	bucketServer := &http.Server{
+		Addr:    bucketAddr,
+		Handler: bucketHandler,
+	}
+	go bucketServer.Serve(bucketListener)
+	defer bucketServer.Shutdown(context.Background())
+
+	bucket := &sourcev1.Bucket{
+		Spec: sourcev1.BucketSpec{
+			Endpoint: bucketEndpoint,
+		},
+	}
+	client, err := NewClient(t.Context(),
+		bucket,
+		withoutCredentials(),
+		withoutRetries())
+	g.Expect(err).ToNot(HaveOccurred())
+
+	visited := map[string]string{}
+	err = client.VisitObjects(t.Context(), bucketName, "", func(path, etag string) error {
+		visited[path] = etag
+		return nil
+	})
+	g.Expect(err).ToNot(HaveOccurred())
+	// The nameless entry is skipped, entries without an etag are
+	// visited with an empty etag.
+	g.Expect(visited).To(Equal(map[string]string{
+		"no-properties.txt": "",
+		"no-etag.txt":       "",
+	}))
+}
+
 func Test_chainCredentialWithSecret(t *testing.T) {
 	g := NewWithT(t)
 

internal/controller/bucket_controller.go

@@ -686,7 +686,7 @@ func fetchEtagIndex(ctx context.Context, provider BucketProvider, obj *sourcev1.
 	// Confirm bucket exists
 	exists, err := provider.BucketExists(ctxTimeout, obj.Spec.BucketName)
 	if err != nil {
-		return fmt.Errorf("failed to confirm existence of '%s' bucket: %w", obj.Spec.BucketName, err)
+		return fmt.Errorf("failed to confirm existence of '%s' bucket: %w", obj.Spec.BucketName, serror.SanitizeError(err))
 	}
 	if !exists {
 		err = fmt.Errorf("bucket '%s' not found", obj.Spec.BucketName)
@@ -724,7 +724,7 @@ func fetchEtagIndex(ctx context.Context, provider BucketProvider, obj *sourcev1.
 		return nil
 	})
 	if err != nil {
-		return fmt.Errorf("indexation of objects from bucket '%s' failed: %w", obj.Spec.BucketName, err)
+		return fmt.Errorf("indexation of objects from bucket '%s' failed: %w", obj.Spec.BucketName, serror.SanitizeError(err))
 	}
 	return nil
 }
@@ -751,7 +751,12 @@ func fetchIndexFiles(ctx context.Context, provider BucketProvider, obj *sourcev1
 			if err := sem.Acquire(groupCtx, 1); err != nil {
 				return err
 			}
-			group.Go(func() error {
+			group.Go(func() (err error) {
+				defer func() {
+					if r := recover(); r != nil {
+						err = fmt.Errorf("failed to get '%s' object: %v", k, r)
+					}
+				}()
 				defer sem.Release(1)
 				localPath, err := securejoin.SecureJoin(tempDir, k)
 				if err != nil {

internal/controller/bucket_controller_fetch_test.go

@@ -38,13 +38,18 @@ type mockBucketObject struct {
 }
 
 type mockBucketClient struct {
-	bucketName string
-	objects    map[string]mockBucketObject
+	bucketName      string
+	objects         map[string]mockBucketObject
+	bucketExistsErr error
+	visitErr        error
 }
 
 var errMockNotFound = fmt.Errorf("not found")
 
 func (m mockBucketClient) BucketExists(_ context.Context, name string) (bool, error) {
+	if m.bucketExistsErr != nil {
+		return false, m.bucketExistsErr
+	}
 	return name == m.bucketName, nil
 }
 
@@ -71,6 +76,9 @@ func (m mockBucketClient) ObjectIsNotFound(e error) bool {
 }
 
 func (m mockBucketClient) VisitObjects(_ context.Context, _ string, _ string, f func(key, etag string) error) error {
+	if m.visitErr != nil {
+		return m.visitErr
+	}
 	for key, obj := range m.objects {
 		if err := f(key, obj.etag); err != nil {
 			return err
@@ -136,6 +144,38 @@ func Test_fetchEtagIndex(t *testing.T) {
 		g.Expect(err.Error()).To(ContainSubstring("not found"))
 	})
 
+	t.Run("sanitizes bucket existence error", func(t *testing.T) {
+		g := NewWithT(t)
+		tmp := t.TempDir()
+
+		client := mockBucketClient{
+			bucketName:      bucketName,
+			bucketExistsErr: fmt.Errorf(`Get "https://account.blob.core.windows.net/container?comp=list&sv=2022-11-02&sig=credential": dial tcp: connection refused`),
+		}
+
+		index := index.NewDigester()
+		err := fetchEtagIndex(context.TODO(), client, bucket.DeepCopy(), index, tmp)
+		g.Expect(err).To(HaveOccurred())
+		g.Expect(err.Error()).To(ContainSubstring("https://account.blob.core.windows.net/container"))
+		g.Expect(err.Error()).ToNot(ContainSubstring("sig="))
+	})
+
+	t.Run("sanitizes object listing error", func(t *testing.T) {
+		g := NewWithT(t)
+		tmp := t.TempDir()
+
+		client := mockBucketClient{
+			bucketName: bucketName,
+			visitErr:   fmt.Errorf(`Get "https://account.blob.core.windows.net/container?comp=list&sv=2022-11-02&sig=credential": read: connection reset by peer`),
+		}
+
+		index := index.NewDigester()
+		err := fetchEtagIndex(context.TODO(), client, bucket.DeepCopy(), index, tmp)
+		g.Expect(err).To(HaveOccurred())
+		g.Expect(err.Error()).To(ContainSubstring("https://account.blob.core.windows.net/container"))
+		g.Expect(err.Error()).ToNot(ContainSubstring("sig="))
+	})
+
 	t.Run("filters with .sourceignore rules", func(t *testing.T) {
 		tmp := t.TempDir()