Path Traversal

diff --git a/pkg/storage/unified/search/bleve.go b/pkg/storage/unified/search/bleve.go index 5bb97aed065a9..f373d98add1e6 100644 --- a/pkg/storage/unified/search/bleve.go +++ b/pkg/storage/unified/search/bleve.go @@ -622,7 +622,13 @@ func (b *bleveBackend) BuildIndex( } func (b *bleveBackend) getResourceDir(key resource.NamespacedResource) string { - return filepath.Join(b.opts.Root, cleanFileSegment(key.Namespace), cleanFileSegment(fmt.Sprintf("%s.%s", key.Resource, key.Group))) + return filepath.Join(b.opts.Root, resourceSubPath(key)) +} + +// resourceSubPath returns the namespaced on-disk/object-store path for a resource, +// for example: default/dashboards.dashboard.grafana.app +func resourceSubPath(key resource.NamespacedResource) string { + return filepath.Join(cleanFileSegment(key.Namespace), cleanFileSegment(fmt.Sprintf("%s.%s", key.Resource, key.Group))) } func cleanFileSegment(input string) string { diff --git a/pkg/storage/unified/search/bleve_snapshot_upload.go b/pkg/storage/unified/search/bleve_snapshot_upload.go new file mode 100644 index 0000000000000..a50a6bc5ba630 --- /dev/null +++ b/pkg/storage/unified/search/bleve_snapshot_upload.go @@ -0,0 +1,116 @@ +package search + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/blevesearch/bleve/v2" + + "github.com/grafana/grafana/pkg/storage/unified/resource" +) + +func (b *bleveBackend) uploadSnapshot(ctx context.Context, key resource.NamespacedResource, idx *bleveIndex) (retErr error) { + lock, err := b.opts.Snapshot.Store.LockBuildIndex(ctx, key) + if err != nil { + return fmt.Errorf("acquiring snapshot upload lock: %w", err) + } + defer func() { + releaseErr := lock.Release() + if releaseErr == nil { + return + } + if retErr == nil { + retErr = fmt.Errorf("releasing snapshot upload lock: %w", releaseErr) + return + } + retErr = errors.Join(retErr, fmt.Errorf("releasing snapshot upload lock: %w", releaseErr)) + }() + + stagingDir, err := b.newSnapshotStagingDir(key) + if err != nil { + return fmt.Errorf("creating snapshot staging dir: %w", err) + } + defer func() { _ = os.RemoveAll(stagingDir) }() + + start := time.Now() + if err := b.snapshotIndex(idx.index, stagingDir); err != nil { + return err + } + // Lock loss is checked only at step boundaries. The main value of the + // distributed lock is preventing duplicate snapshot/upload work up front, and + // the upload path is safe to retry because remote snapshots are immutable and + // keyed by unique ULIDs. + if err := checkSnapshotLock(lock); err != nil { + return err + } + + // Read RV/build info from the staged snapshot instead of the live index so + // the uploaded metadata matches the copied snapshot contents even if the live + // index advanced while CopyTo was running. + snapshotIdx, err := bleve.OpenUsing(stagingDir, map[string]interface{}{"bolt_timeout": boltTimeout}) + if err != nil { + return fmt.Errorf("opening staged snapshot: %w", err) + } + + rv, rvErr := getRV(snapshotIdx) + bi, biErr := getBuildInfo(snapshotIdx) + if closeErr := snapshotIdx.Close(); closeErr != nil { + return fmt.Errorf("closing staged snapshot: %w", closeErr) + } + if rvErr != nil { + return fmt.Errorf("reading snapshot rv: %w", rvErr) + } + if biErr != nil { + return fmt.Errorf("reading snapshot build info: %w", biErr) + } + + _, err = b.opts.Snapshot.Store.UploadIndex(ctx, key, stagingDir, IndexMeta{ + GrafanaBuildVersion: bi.BuildVersion, + LatestResourceVersion: rv, + }) + if err != nil { + return fmt.Errorf("uploading snapshot: %w", err) + } + if err := checkSnapshotLock(lock); err != nil { + return err + } + + elapsed := time.Since(start) + b.log.Info("Uploaded remote index snapshot", "key", key, "elapsed", elapsed, "rv", rv) + return nil +} + +func (b *bleveBackend) snapshotIndex(idx bleve.Index, destDir string) error { + copyable, ok := idx.(bleve.IndexCopyable) + if !ok { + return fmt.Errorf("index does not support snapshot copy") + } + if err := copyable.CopyTo(bleve.FileSystemDirectory(destDir)); err != nil { + return fmt.Errorf("copying index snapshot: %w", err) + } + return nil +} + +func (b *bleveBackend) newSnapshotStagingDir(key resource.NamespacedResource) (string, error) { + parent := filepath.Join(b.opts.Root, "snapshots", resourceSubPath(key)) + if !isPathWithinRoot(parent, b.opts.Root) { + return "", fmt.Errorf("invalid path %s", parent) + } + if err := os.MkdirAll(parent, 0o700); err != nil { + return "", err + } + return os.MkdirTemp(parent, "upload-*") +} + +func checkSnapshotLock(lock IndexStoreLock) error { + select { + case <-lock.Lost(): + return fmt.Errorf("snapshot upload lock lost") + default: + return nil + } +} diff --git a/pkg/storage/unified/search/bleve_snapshot_upload_test.go b/pkg/storage/unified/search/bleve_snapshot_upload_test.go new file mode 100644 index 0000000000000..194ee50ca3d59 --- /dev/null +++ b/pkg/storage/unified/search/bleve_snapshot_upload_test.go @@ -0,0 +1,160 @@ +package search + +import ( + "context" + "errors" + "io/fs" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/blevesearch/bleve/v2" + "github.com/oklog/ulid/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/grafana/grafana/pkg/storage/unified/resource" +) + +type uploadTestStore struct { + lockErr error + + uploadErr error + uploadCalls atomic.Int32 + uploadMeta IndexMeta + uploaded []string +} + +func (s *uploadTestStore) LockBuildIndex(context.Context, resource.NamespacedResource) (IndexStoreLock, error) { + if s.lockErr != nil { + return nil, s.lockErr + } + return &noopIndexStoreLock{lost: make(chan struct{})}, nil +} + +func (s *uploadTestStore) UploadIndex(_ context.Context, _ resource.NamespacedResource, localDir string, meta IndexMeta) (ulid.ULID, error) { + s.uploadCalls.Add(1) + s.uploadMeta = meta + + err := filepath.WalkDir(localDir, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + return nil + } + rel, err := filepath.Rel(localDir, path) + if err != nil { + return err + } + s.uploaded = append(s.uploaded, filepath.ToSlash(rel)) + return nil + }) + if err != nil { + return ulid.ULID{}, err + } + if s.uploadErr != nil { + return ulid.ULID{}, s.uploadErr + } + return ulid.Make(), nil +} + +func (s *uploadTestStore) DownloadIndex(context.Context, resource.NamespacedResource, ulid.ULID, string) (*IndexMeta, error) { + panic("DownloadIndex not implemented for uploadTestStore") +} + +func (s *uploadTestStore) ListIndexes(context.Context, resource.NamespacedResource) (map[ulid.ULID]*IndexMeta, error) { + panic("ListIndexes not implemented for uploadTestStore") +} + +func (s *uploadTestStore) DeleteIndex(context.Context, resource.NamespacedResource, ulid.ULID) error { + panic("DeleteIndex not implemented for uploadTestStore") +} + +func (s *uploadTestStore) CleanupIncompleteUploads(context.Context, resource.NamespacedResource, time.Duration) (int, error) { + panic("CleanupIncompleteUploads not implemented for uploadTestStore") +} + +func newUploadTestIndex(t *testing.T, be *bleveBackend, key resource.NamespacedResource, rv int64) *bleveIndex { + t.Helper() + resourceDir := be.getResourceDir(key) + require.NoError(t, os.MkdirAll(resourceDir, 0o750)) + + idx, err := newBleveIndex(filepath.Join(resourceDir, formatIndexName(time.Now())), bleve.NewIndexMapping(), time.Now(), be.opts.BuildVersion, nil) + require.NoError(t, err) + t.Cleanup(func() { _ = idx.Close() }) + + require.NoError(t, idx.Index("dash-1", map[string]string{"title": "Production Overview"})) + require.NoError(t, setRV(idx, rv)) + + wrapped := be.newBleveIndex(key, idx, indexStorageFile, nil, nil, nil, nil, be.log) + wrapped.resourceVersion.Store(rv) + return wrapped +} + +func TestSnapshotIndex_CreatesUsableCopy(t *testing.T) { + be, _ := newTestBleveBackend(t, SnapshotOptions{}) + key := newTestNsResource() + src := newUploadTestIndex(t, be, key, 42) + + destDir := filepath.Join(t.TempDir(), "snapshot") + require.NoError(t, be.snapshotIndex(src.index, destDir)) + + copied, err := bleve.OpenUsing(destDir, map[string]interface{}{"bolt_timeout": boltTimeout}) + require.NoError(t, err) + defer func() { _ = copied.Close() }() + + rv, err := getRV(copied) + require.NoError(t, err) + assert.Equal(t, int64(42), rv) + + count, err := copied.DocCount() + require.NoError(t, err) + assert.Equal(t, uint64(1), count) +} + +func TestUploadSnapshot_Success(t *testing.T) { + store := &uploadTestStore{} + be, _ := newTestBleveBackend(t, SnapshotOptions{Store: store}) + key := newTestNsResource() + idx := newUploadTestIndex(t, be, key, 42) + + require.NoError(t, be.uploadSnapshot(context.Background(), key, idx)) + assert.Equal(t, int32(1), store.uploadCalls.Load()) + assert.Equal(t, int64(42), store.uploadMeta.LatestResourceVersion) + assert.Equal(t, be.opts.BuildVersion, store.uploadMeta.GrafanaBuildVersion) + assert.NotEmpty(t, store.uploaded) + + snapshotParent := filepath.Join(be.opts.Root, "snapshots", resourceSubPath(key)) + entries, err := os.ReadDir(snapshotParent) + require.NoError(t, err) + assert.Empty(t, entries) +} + +func TestUploadSnapshot_LockContention(t *testing.T) { + store := &uploadTestStore{lockErr: errLockHeld} + be, _ := newTestBleveBackend(t, SnapshotOptions{Store: store}) + key := newTestNsResource() + idx := newUploadTestIndex(t, be, key, 42) + + err := be.uploadSnapshot(context.Background(), key, idx) + require.ErrorIs(t, err, errLockHeld) + assert.Zero(t, store.uploadCalls.Load()) +} + +func TestUploadSnapshot_UploadErrorCleansStagingDir(t *testing.T) { + store := &uploadTestStore{uploadErr: errors.New("boom")} + be, _ := newTestBleveBackend(t, SnapshotOptions{Store: store}) + key := newTestNsResource() + idx := newUploadTestIndex(t, be, key, 42) + + err := be.uploadSnapshot(context.Background(), key, idx) + require.Error(t, err) + + snapshotParent := filepath.Join(be.opts.Root, "snapshots", resourceSubPath(key)) + entries, readErr := os.ReadDir(snapshotParent) + require.NoError(t, readErr) + assert.Empty(t, entries) +} diff --git a/pkg/storage/unified/search/remote_index_store.go b/pkg/storage/unified/search/remote_index_store.go index c722c60603ea6..1557360f0e134 100644 --- a/pkg/storage/unified/search/remote_index_store.go +++ b/pkg/storage/unified/search/remote_index_store.go @@ -109,16 +109,16 @@ func NewBucketRemoteIndexStore(bucket resource.CDKBucket, lockBackend lockBacken // indexPrefix returns the object storage prefix for a namespaced resource + index key. func indexPrefix(ns resource.NamespacedResource, indexKey string) string { - return fmt.Sprintf("%s/%s.%s/%s/", ns.Namespace, ns.Group, ns.Resource, indexKey) + return fmt.Sprintf("%s/%s/", resourceSubPath(ns), indexKey) } // nsPrefix returns the object storage prefix for a namespaced resource (without index key). func nsPrefix(ns resource.NamespacedResource) string { - return fmt.Sprintf("%s/%s.%s/", ns.Namespace, ns.Group, ns.Resource) + return fmt.Sprintf("%s/", resourceSubPath(ns)) } func buildIndexLockKey(ns resource.NamespacedResource) string { - return fmt.Sprintf("%s/%s.%s/locks/build", ns.Namespace, ns.Group, ns.Resource) + return fmt.Sprintf("%s/locks/build", resourceSubPath(ns)) } func (s *BucketRemoteIndexStore) LockBuildIndex(ctx context.Context, nsResource resource.NamespacedResource) (IndexStoreLock, error) {