From 9afd87144fcbf147e29451fa22c2b45b0e548c61 Mon Sep 17 00:00:00 2001 From: Victor Chen Date: Mon, 10 Aug 2026 22:03:25 +1000 Subject: [PATCH 1/2] fix(xpkg): prevent corrupt package cache entries Signed-off-by: Victor Chen --- pkg/xpkg/cache.go | 53 ++++++++- pkg/xpkg/cache_test.go | 177 +++++++++++++++++++++++++++++ pkg/xpkg/client.go | 80 ++++++++----- pkg/xpkg/client_test.go | 243 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 520 insertions(+), 33 deletions(-) diff --git a/pkg/xpkg/cache.go b/pkg/xpkg/cache.go index a23614fce..def4c0b9f 100644 --- a/pkg/xpkg/cache.go +++ b/pkg/xpkg/cache.go @@ -49,6 +49,23 @@ type FsPackageCache struct { mu sync.RWMutex } +type unlockingReadCloser struct { + io.ReadCloser + + once sync.Once + err error + unlock func() +} + +func (r *unlockingReadCloser) Close() error { + r.once.Do(func() { + r.err = r.ReadCloser.Close() + r.unlock() + }) + + return r.err +} + // NewFsPackageCache creates a new FsPackageCache. func NewFsPackageCache(dir string, fs afero.Fs) *FsPackageCache { return &FsPackageCache{ @@ -59,6 +76,9 @@ func NewFsPackageCache(dir string, fs afero.Fs) *FsPackageCache { // Has indicates whether an item with the given id is in the cache. func (c *FsPackageCache) Has(id string) bool { + c.mu.RLock() + defer c.mu.RUnlock() + if fi, err := c.fs.Stat(BuildPath(c.dir, id, cacheContentExt)); err == nil && !fi.IsDir() { return true } @@ -69,14 +89,24 @@ func (c *FsPackageCache) Has(id string) bool { // Get retrieves package contents from the cache. func (c *FsPackageCache) Get(id string) (io.ReadCloser, error) { c.mu.RLock() - defer c.mu.RUnlock() f, err := c.fs.Open(BuildPath(c.dir, id, cacheContentExt)) if err != nil { + c.mu.RUnlock() return nil, err } - return GzipReadCloser(f) + r, err := GzipReadCloser(f) + if err != nil { + _ = f.Close() + c.mu.RUnlock() + return nil, err + } + + return &unlockingReadCloser{ + ReadCloser: r, + unlock: c.mu.RUnlock, + }, nil } // Store saves the package contents to the cache. @@ -84,28 +114,41 @@ func (c *FsPackageCache) Store(id string, content io.ReadCloser) error { c.mu.Lock() defer c.mu.Unlock() - cf, err := c.fs.Create(BuildPath(c.dir, id, cacheContentExt)) + path := BuildPath(c.dir, id, cacheContentExt) + cf, err := c.fs.Create(path) if err != nil { return err } - defer cf.Close() //nolint:errcheck // Error is checked in the happy path. + cleanup := func() { + _ = cf.Close() + _ = c.fs.Remove(path) + } w, err := gzip.NewWriterLevel(cf, gzip.BestSpeed) if err != nil { + cleanup() return err } _, err = io.Copy(w, content) if err != nil { + _ = w.Close() + cleanup() return err } // NOTE(hasheddan): gzip writer must be closed to ensure all data is flushed // to file. if err := w.Close(); err != nil { + cleanup() return err } - return cf.Close() + if err := cf.Close(); err != nil { + cleanup() + return err + } + + return nil } // Delete removes package contents from the cache. diff --git a/pkg/xpkg/cache_test.go b/pkg/xpkg/cache_test.go index db92ef402..07eca200c 100644 --- a/pkg/xpkg/cache_test.go +++ b/pkg/xpkg/cache_test.go @@ -19,10 +19,14 @@ package xpkg import ( "bytes" "compress/gzip" + "errors" "io" "os" + "strings" "syscall" "testing" + "testing/iotest" + "time" "github.com/google/go-cmp/cmp" "github.com/spf13/afero" @@ -32,6 +36,44 @@ import ( var _ PackageCache = &FsPackageCache{} +type errorFile struct { + afero.File + writeErr error + closeErr error +} + +func (f *errorFile) Write(p []byte) (int, error) { + if f.writeErr != nil { + return 0, f.writeErr + } + + return f.File.Write(p) +} + +func (f *errorFile) Close() error { + err := f.File.Close() + if f.closeErr != nil { + return f.closeErr + } + + return err +} + +type errorFs struct { + afero.Fs + writeErr error + closeErr error +} + +func (f *errorFs) Create(name string) (afero.File, error) { + file, err := f.Fs.Create(name) + if err != nil { + return nil, err + } + + return &errorFile{File: file, writeErr: f.writeErr, closeErr: f.closeErr}, nil +} + func TestHas(t *testing.T) { fs := afero.NewMemMapFs() cf, _ := fs.Create("/cache/exists.gz") @@ -181,6 +223,141 @@ func TestStore(t *testing.T) { } } +func TestStoreRoundTrip(t *testing.T) { + cache := NewFsPackageCache("/cache", afero.NewMemMapFs()) + want := "package content" + + if err := cache.Store("package", io.NopCloser(strings.NewReader(want))); err != nil { + t.Fatalf("Store(...): unexpected error: %v", err) + } + + r, err := cache.Get("package") + if err != nil { + t.Fatalf("Get(...): unexpected error: %v", err) + } + defer r.Close() + + got, err := io.ReadAll(r) + if err != nil { + t.Fatalf("Read(...): unexpected error: %v", err) + } + if diff := cmp.Diff(want, string(got)); diff != "" { + t.Errorf("Store(...): -want content, +got content:\n%s", diff) + } +} + +func TestStoreRemovesFailedWrites(t *testing.T) { + errWrite := errors.New("write failed") + cases := map[string]struct { + fs afero.Fs + content io.ReadCloser + }{ + "ContentReadError": { + fs: afero.NewMemMapFs(), + content: io.NopCloser(io.MultiReader(strings.NewReader("partial"), iotest.ErrReader(errWrite))), + }, + "GzipCloseError": { + fs: &errorFs{Fs: afero.NewMemMapFs(), writeErr: errWrite}, + content: io.NopCloser(strings.NewReader("")), + }, + "FileCloseError": { + fs: &errorFs{Fs: afero.NewMemMapFs(), closeErr: errWrite}, + content: io.NopCloser(strings.NewReader("content")), + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + cache := NewFsPackageCache("/cache", tc.fs) + if err := cache.Store("package", tc.content); err == nil { + t.Fatal("Store(...): expected an error") + } + if cache.Has("package") { + t.Fatal("Store(...): failed write remained in cache") + } + }) + } +} + +func TestStoreWaitsForReader(t *testing.T) { + cache := NewFsPackageCache("/cache", afero.NewMemMapFs()) + if err := cache.Store("package", io.NopCloser(strings.NewReader("old"))); err != nil { + t.Fatalf("Store(...): unexpected error: %v", err) + } + + r, err := cache.Get("package") + if err != nil { + t.Fatalf("Get(...): unexpected error: %v", err) + } + + started := make(chan struct{}) + done := make(chan error, 1) + go func() { + close(started) + done <- cache.Store("package", io.NopCloser(strings.NewReader("new"))) + }() + <-started + + select { + case err := <-done: + t.Fatalf("Store(...) completed while cache reader was open: %v", err) + case <-time.After(100 * time.Millisecond): + } + + got, err := io.ReadAll(r) + if err != nil { + t.Fatalf("Read(...): unexpected error: %v", err) + } + if diff := cmp.Diff("old", string(got)); diff != "" { + t.Errorf("Get(...): -want content, +got content:\n%s", diff) + } + if err := r.Close(); err != nil { + t.Fatalf("Close(...): unexpected error: %v", err) + } + + select { + case err := <-done: + if err != nil { + t.Fatalf("Store(...): unexpected error: %v", err) + } + case <-time.After(time.Second): + t.Fatal("Store(...) did not complete after cache reader closed") + } +} + +func TestGetErrorReleasesLock(t *testing.T) { + fs := afero.NewMemMapFs() + f, err := fs.Create("/cache/package.gz") + if err != nil { + t.Fatalf("Create(...): unexpected error: %v", err) + } + if _, err := f.WriteString("not gzip"); err != nil { + t.Fatalf("WriteString(...): unexpected error: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("Close(...): unexpected error: %v", err) + } + + cache := NewFsPackageCache("/cache", fs) + if _, err := cache.Get("package"); err == nil { + t.Fatal("Get(...): expected an error") + } + + done := make(chan error, 1) + go func() { + done <- cache.Delete("package") + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("Delete(...): unexpected error: %v", err) + } + case <-time.After(time.Second): + t.Fatal("Delete(...) blocked after Get(...) failed") + } +} + func TestDelete(t *testing.T) { fs := afero.NewMemMapFs() _, _ = fs.Create("/cache/exists.xpkg") diff --git a/pkg/xpkg/client.go b/pkg/xpkg/client.go index 656f83825..0e624d8c0 100644 --- a/pkg/xpkg/client.go +++ b/pkg/xpkg/client.go @@ -217,6 +217,24 @@ func NewCachedClient(f Fetcher, p parser.Parser, c PackageCache, s ConfigStore, } } +func (c *CachedClient) getCachedPackage(ctx context.Context, key string) (*parser.Package, bool, error) { + rc, err := c.cache.Get(key) + if err != nil { + return nil, false, err + } + + pkg, err := c.parser.Parse(ctx, struct { + io.Reader + io.Closer + }{ + Reader: io.LimitReader(rc, maxPackageSize), + Closer: rc, + }) + _ = rc.Close() + + return pkg, true, err +} + // Get fetches and parses a complete package. func (c *CachedClient) Get(ctx context.Context, ref string, opts ...GetOption) (*Package, error) { cfg := &GetConfig{ @@ -275,32 +293,26 @@ func (c *CachedClient) Get(ctx context.Context, ref string, opts ...GetOption) ( cacheKey := FriendlyID(ParsePackageSourceFromReference(parsedOriginalRef), digest) if cfg.pullPolicy != corev1.PullAlways { - rc, err := c.cache.Get(cacheKey) + pkg, found, err := c.getCachedPackage(ctx, cacheKey) if err == nil { - pkg, err := c.parser.Parse(ctx, struct { - io.Reader - io.Closer - }{ - Reader: io.LimitReader(rc, maxPackageSize), - Closer: rc, - }) - rc.Close() //nolint:errcheck // Only open for reading. - if err == nil { - return &Package{ - Package: pkg, - Digest: digest, - Version: parsedOriginalRef.Identifier(), - Source: ParsePackageSourceFromReference(parsedOriginalRef), - ResolvedVersion: parsedResolvedRef.Identifier(), - ResolvedSource: ParsePackageSourceFromReference(parsedResolvedRef), - AppliedImageConfigs: applied, - }, nil - } + return &Package{ + Package: pkg, + Digest: digest, + Version: parsedOriginalRef.Identifier(), + Source: ParsePackageSourceFromReference(parsedOriginalRef), + ResolvedVersion: parsedResolvedRef.Identifier(), + ResolvedSource: ParsePackageSourceFromReference(parsedResolvedRef), + AppliedImageConfigs: applied, + }, nil } - } - if cfg.pullPolicy == corev1.PullNever { - return nil, errors.New("package not in cache and pull policy is Never") + if cfg.pullPolicy == corev1.PullNever { + return nil, errors.Wrapf(err, "cannot use cached package %s and pull policy is Never", resolvedRef) + } + + if found { + _ = c.cache.Delete(cacheKey) + } } // Verification only happens if we don't get a cache hit. This means we @@ -342,24 +354,36 @@ func (c *CachedClient) Get(ctx context.Context, ref string, opts ...GetOption) ( } pipeR, pipeW := io.Pipe() - teeRC := TeeReadCloser(rc, pipeW) - defer teeRC.Close() //nolint:errcheck // Would only error if we called pipeW.CloseWithError() + cacheWrite := make(chan error, 1) go func() { defer pipeR.Close() //nolint:errcheck // Only open for reading. - _ = c.cache.Store(cacheKey, pipeR) + err := c.cache.Store(cacheKey, pipeR) + _, _ = io.Copy(io.Discard, pipeR) + cacheWrite <- err }() pkg, err := c.parser.Parse(ctx, struct { io.Reader io.Closer }{ - Reader: io.LimitReader(teeRC, maxPackageSize), - Closer: teeRC, + Reader: io.LimitReader(io.TeeReader(rc, pipeW), maxPackageSize), + Closer: rc, }) if err != nil { + _ = pipeW.CloseWithError(err) + } else { + _ = pipeW.Close() + } + + cacheErr := <-cacheWrite + if err != nil { + _ = c.cache.Delete(cacheKey) return nil, errors.Wrapf(err, "cannot parse package %s", resolvedRef) } + if cacheErr != nil { + _ = c.cache.Delete(cacheKey) + } return &Package{ Package: pkg, diff --git a/pkg/xpkg/client_test.go b/pkg/xpkg/client_test.go index a1e7db8cd..5fca671e7 100644 --- a/pkg/xpkg/client_test.go +++ b/pkg/xpkg/client_test.go @@ -23,12 +23,14 @@ import ( "io" "strings" "testing" + "time" "github.com/crossplane/crossplane/apis/v2/pkg/v1beta1" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" "github.com/google/go-containerregistry/pkg/name" v1 "github.com/google/go-containerregistry/pkg/v1" + "github.com/spf13/afero" corev1 "k8s.io/api/core/v1" "github.com/crossplane/crossplane-runtime/v2/pkg/errors" @@ -79,6 +81,10 @@ func (m *MockCache) Store(key string, rc io.ReadCloser) error { } func (m *MockCache) Delete(key string) error { + if m.MockDelete == nil { + return nil + } + return m.MockDelete(key) } @@ -203,6 +209,62 @@ func NewTestPackage(t *testing.T, metaJSON string, objectsJSON ...string) *parse return pkg } +func NewTestClientForPackage(t *testing.T, packageYAML string) *CachedClient { + t.Helper() + + tarContent := CreateTarWithPackageYAML(packageYAML) + return &CachedClient{ + fetcher: &MockFetcher{ + MockHead: func(_ context.Context, _ name.Reference, _ ...string) (*v1.Descriptor, error) { + return &v1.Descriptor{ + Digest: v1.Hash{ + Algorithm: "sha256", + Hex: "abc123def456789012345678901234567890123456789012345678901234abcd", + }, + }, nil + }, + MockFetch: func(_ context.Context, _ name.Reference, _ ...string) (v1.Image, error) { + return &MockImage{ + MockManifest: func() (*v1.Manifest, error) { + return &v1.Manifest{ + Layers: []v1.Descriptor{ + { + Annotations: map[string]string{AnnotationKey: PackageAnnotation}, + Digest: v1.Hash{Algorithm: "sha256", Hex: "layer123"}, + }, + }, + }, nil + }, + MockLayerByDigest: func(_ v1.Hash) (v1.Layer, error) { + return NewMockLayer(tarContent), nil + }, + }, nil + }, + }, + parser: NewTestParser(t), + cache: &MockCache{ + MockGet: func(_ string) (io.ReadCloser, error) { + return nil, errors.New("not in cache") + }, + MockStore: func(_ string, rc io.ReadCloser) error { + _, _ = io.Copy(io.Discard, rc) + return nil + }, + }, + config: &MockConfigStore{ + MockRewritePath: func(_ context.Context, _ string) (string, string, error) { + return "", "", nil + }, + MockPullSecretFor: func(_ context.Context, _ string) (string, string, error) { + return "", "", nil + }, + MockImageVerificationConfigFor: func(_ context.Context, _ string) (string, *v1beta1.ImageVerification, error) { + return "", nil, nil + }, + }, + } +} + func PackageComparer() cmp.Option { return cmp.Comparer(func(a, b *parser.Package) bool { if a == nil && b == nil { @@ -1057,6 +1119,187 @@ func TestClientGet(t *testing.T) { } } +func TestClientGetWaitsForCacheStore(t *testing.T) { + providerMeta := `{"apiVersion":"meta.pkg.crossplane.io/v1","kind":"Provider","metadata":{"name":"provider-aws"}}` + client := NewTestClientForPackage(t, providerMeta) + stored := make(chan struct{}) + release := make(chan struct{}) + client.cache = &MockCache{ + MockGet: func(_ string) (io.ReadCloser, error) { + return nil, errors.New("not in cache") + }, + MockStore: func(_ string, rc io.ReadCloser) error { + _, _ = io.Copy(io.Discard, rc) + close(stored) + <-release + return nil + }, + } + + done := make(chan error, 1) + go func() { + _, err := client.Get(context.Background(), testSource+":"+testTag) + done <- err + }() + <-stored + + select { + case err := <-done: + t.Fatalf("Get(...) returned before cache Store(...) completed: %v", err) + case <-time.After(100 * time.Millisecond): + } + + close(release) + select { + case err := <-done: + if err != nil { + t.Fatalf("Get(...): unexpected error: %v", err) + } + case <-time.After(time.Second): + t.Fatal("Get(...) did not return after cache Store(...) completed") + } +} + +func TestClientGetIgnoresCacheStoreFailure(t *testing.T) { + providerMeta := `{"apiVersion":"meta.pkg.crossplane.io/v1","kind":"Provider","metadata":{"name":"provider-aws"}}` + client := NewTestClientForPackage(t, providerMeta) + deleted := 0 + client.cache = &MockCache{ + MockGet: func(_ string) (io.ReadCloser, error) { + return nil, errors.New("not in cache") + }, + MockStore: func(_ string, _ io.ReadCloser) error { + return errors.New("cache full") + }, + MockDelete: func(_ string) error { + deleted++ + return nil + }, + } + + pkg, err := client.Get(context.Background(), testSource+":"+testTag) + if err != nil { + t.Fatalf("Get(...): unexpected error: %v", err) + } + if pkg == nil { + t.Fatal("Get(...): expected a package") + } + if deleted != 1 { + t.Errorf("Delete(...): want 1 call, got %d", deleted) + } +} + +func TestClientGetRefetchesCorruptCache(t *testing.T) { + providerMeta := `{"apiVersion":"meta.pkg.crossplane.io/v1","kind":"Provider","metadata":{"name":"provider-aws"}}` + client := NewTestClientForPackage(t, providerMeta) + deleted := 0 + client.cache = &MockCache{ + MockGet: func(_ string) (io.ReadCloser, error) { + return io.NopCloser(strings.NewReader("invalid yaml content {{{")), nil + }, + MockStore: func(_ string, rc io.ReadCloser) error { + _, _ = io.Copy(io.Discard, rc) + return nil + }, + MockDelete: func(_ string) error { + deleted++ + return nil + }, + } + + pkg, err := client.Get(context.Background(), testSource+":"+testTag, WithPullPolicy(corev1.PullIfNotPresent)) + if err != nil { + t.Fatalf("Get(...): unexpected error: %v", err) + } + if pkg == nil { + t.Fatal("Get(...): expected a package") + } + if deleted != 1 { + t.Errorf("Delete(...): want 1 call, got %d", deleted) + } +} + +func TestClientGetPreservesCorruptCacheWithPullNever(t *testing.T) { + providerMeta := `{"apiVersion":"meta.pkg.crossplane.io/v1","kind":"Provider","metadata":{"name":"provider-aws"}}` + client := NewTestClientForPackage(t, providerMeta) + head := false + client.fetcher.(*MockFetcher).MockHead = func(_ context.Context, _ name.Reference, _ ...string) (*v1.Descriptor, error) { + head = true + return &v1.Descriptor{Digest: v1.Hash{Algorithm: "sha256", Hex: strings.TrimPrefix(testDigest, "sha256:")}}, nil + } + fetched := false + client.fetcher.(*MockFetcher).MockFetch = func(_ context.Context, _ name.Reference, _ ...string) (v1.Image, error) { + fetched = true + return nil, errors.New("Fetch should not be called with PullNever") + } + deleted := false + client.cache = &MockCache{ + MockGet: func(_ string) (io.ReadCloser, error) { + return io.NopCloser(strings.NewReader("invalid yaml content {{{")), nil + }, + MockDelete: func(_ string) error { + deleted = true + return nil + }, + } + + _, err := client.Get(context.Background(), testSource+":"+testTag, WithPullPolicy(corev1.PullNever)) + if err == nil || !strings.Contains(err.Error(), "cannot use cached package") { + t.Fatalf("Get(...): expected cached package parse error, got %v", err) + } + if deleted { + t.Error("Delete(...): corrupt PullNever cache entry was deleted") + } + if fetched { + t.Error("Fetch(...): registry was called with PullNever") + } + if !head { + t.Error("Head(...): tag reference was not resolved") + } +} + +func TestClientGetDigestWithPullNeverDoesNotUseRegistry(t *testing.T) { + providerMeta := `{"apiVersion":"meta.pkg.crossplane.io/v1","kind":"Provider","metadata":{"name":"provider-aws"}}` + client := NewTestClientForPackage(t, providerMeta) + client.fetcher = &MockFetcher{ + MockHead: func(_ context.Context, _ name.Reference, _ ...string) (*v1.Descriptor, error) { + return nil, errors.New("Head should not be called for digest refs") + }, + MockFetch: func(_ context.Context, _ name.Reference, _ ...string) (v1.Image, error) { + return nil, errors.New("Fetch should not be called with PullNever") + }, + } + client.cache = &MockCache{ + MockGet: func(_ string) (io.ReadCloser, error) { + return io.NopCloser(strings.NewReader(providerMeta)), nil + }, + } + + pkg, err := client.Get(context.Background(), testSource+"@"+testDigest, WithPullPolicy(corev1.PullNever)) + if err != nil { + t.Fatalf("Get(...): unexpected error: %v", err) + } + if pkg == nil { + t.Fatal("Get(...): expected a package") + } +} + +func TestClientGetParseFailureDoesNotRemainCached(t *testing.T) { + fs := afero.NewMemMapFs() + cache := NewFsPackageCache("/cache", fs) + client := NewTestClientForPackage(t, "invalid yaml content {{{") + client.cache = cache + + _, err := client.Get(context.Background(), testSource+":"+testTag) + if err == nil { + t.Fatal("Get(...): expected an error") + } + cacheKey := FriendlyID(testSource, testDigest) + if cache.Has(cacheKey) { + t.Fatal("Get(...): failed package content remained in cache") + } +} + func TestClientListVersions(t *testing.T) { type args struct { source string From c16f7fb6b83b528bb025c7c66e7995ac1c111219 Mon Sep 17 00:00:00 2001 From: Victor Chen Date: Mon, 10 Aug 2026 22:29:34 +1000 Subject: [PATCH 2/2] docs(xpkg): clarify cache reader lifecycle Signed-off-by: Victor Chen --- pkg/xpkg/cache.go | 4 +++- pkg/xpkg/cache_test.go | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/pkg/xpkg/cache.go b/pkg/xpkg/cache.go index def4c0b9f..066879d20 100644 --- a/pkg/xpkg/cache.go +++ b/pkg/xpkg/cache.go @@ -36,6 +36,7 @@ const cacheContentExt = ".gz" // A PackageCache caches package content. type PackageCache interface { Has(id string) bool + // Get returns cached package content. The caller must close the returned reader. Get(id string) (io.ReadCloser, error) Store(id string, content io.ReadCloser) error Delete(id string) error @@ -86,7 +87,8 @@ func (c *FsPackageCache) Has(id string) bool { return false } -// Get retrieves package contents from the cache. +// Get retrieves package contents from the cache. It holds a read lock until the +// returned reader is closed. func (c *FsPackageCache) Get(id string) (io.ReadCloser, error) { c.mu.RLock() diff --git a/pkg/xpkg/cache_test.go b/pkg/xpkg/cache_test.go index 07eca200c..ee69fcb07 100644 --- a/pkg/xpkg/cache_test.go +++ b/pkg/xpkg/cache_test.go @@ -327,7 +327,7 @@ func TestStoreWaitsForReader(t *testing.T) { func TestGetErrorReleasesLock(t *testing.T) { fs := afero.NewMemMapFs() - f, err := fs.Create("/cache/package.gz") + f, err := fs.Create(BuildPath("/cache", "package", cacheContentExt)) if err != nil { t.Fatalf("Create(...): unexpected error: %v", err) }