From ef816e5ea5bab93747f52677d798a91156530e64 Mon Sep 17 00:00:00 2001 From: Stefan Prodan Date: Tue, 4 Aug 2026 21:06:19 +0300 Subject: [PATCH] fix: harden Bucket reconciliation error paths Signed-off-by: Stefan Prodan (cherry picked from commit f6d176b9e60fb19b2ab0d4503efd2a645b4ed604) --- internal/bucket/azure/blob.go | 22 +++- internal/bucket/azure/blob_test.go | 105 ++++++++++++++++++ internal/controller/bucket_controller.go | 11 +- .../bucket_controller_fetch_test.go | 44 +++++++- 4 files changed, 175 insertions(+), 7 deletions(-) diff --git a/internal/bucket/azure/blob.go b/internal/bucket/azure/blob.go index d89b63bb8..94489add9 100644 --- a/internal/bucket/azure/blob.go +++ b/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 } diff --git a/internal/bucket/azure/blob_test.go b/internal/bucket/azure/blob_test.go index 889fe999a..433f0f14c 100644 --- a/internal/bucket/azure/blob_test.go +++ b/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(` + + + + no-properties.txt + + + no-etag.txt + + + + + 0x8D9B2A2A2A2A2A2 + + + + +`, 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) diff --git a/internal/controller/bucket_controller.go b/internal/controller/bucket_controller.go index 78af8f117..797b48709 100644 --- a/internal/controller/bucket_controller.go +++ b/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 { diff --git a/internal/controller/bucket_controller_fetch_test.go b/internal/controller/bucket_controller_fetch_test.go index a0f26a1c9..9f9c31a12 100644 --- a/internal/controller/bucket_controller_fetch_test.go +++ b/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()