private/metaclient: fix multi-passphrase ListObjects paging

If no objects in a list page are able to be decrypted listing stops
prematurely. Use a cursor from the encrypted list instead and page
until at least one object can be decrypted.

Change-Id: I04331e65423c9ff0f5c37b189c2ab467cdcd9b8b

private/metaclient/objects.go

@@ -358,34 +358,54 @@ func (db *DB) ListObjects(ctx context.Context, bucket string, options ListOption
 		return ObjectList{}, errClass.Wrap(err)
 	}
 
-	items, more, err := db.metainfo.ListObjects(ctx, ListObjectsParams{
-		Bucket:                []byte(bucket),
-		EncryptedPrefix:       []byte(pi.ParentEnc.Raw()),
-		EncryptedCursor:       []byte(startAfter),
-		Limit:                 int32(options.Limit),
-		IncludeCustomMetadata: options.IncludeCustomMetadata,
-		IncludeSystemMetadata: options.IncludeSystemMetadata,
-		Recursive:             options.Recursive,
-		Status:                options.Status,
-	})
-	if err != nil {
-		return ObjectList{}, errClass.Wrap(err)
-	}
+	startAfterEnc := []byte(startAfter)
+	if len(options.CursorEnc) > 0 {
+		startAfterEnc = options.CursorEnc
+	}
+
+	var m bool
+	var objectsList []Object
+	// Keep looking until we find an object we can decrypt or we run out of objects
+	for {
+		items, more, err := db.metainfo.ListObjects(ctx, ListObjectsParams{
+			Bucket:                []byte(bucket),
+			EncryptedPrefix:       []byte(pi.ParentEnc.Raw()),
+			EncryptedCursor:       startAfterEnc,
+			Limit:                 int32(options.Limit),
+			IncludeCustomMetadata: options.IncludeCustomMetadata,
+			IncludeSystemMetadata: options.IncludeSystemMetadata,
+			Recursive:             options.Recursive,
+			Status:                options.Status,
+		})
+		if err != nil {
+			return ObjectList{}, errClass.Wrap(err)
+		}
+		m = more
 
-	objectsList, err := db.objectsFromRawObjectList(ctx, items, pi, startAfter)
-	if err != nil {
-		return ObjectList{}, errClass.Wrap(err)
+		objectsList, err = db.objectsFromRawObjectList(ctx, items, pi)
+		if err != nil {
+			return ObjectList{}, errClass.Wrap(err)
+		}
+
+		if len(items) > 0 {
+			startAfterEnc = items[len(items)-1].EncryptedObjectKey
+		}
+
+		if len(objectsList) != 0 || !more {
+			break
+		}
 	}
 
 	return ObjectList{
 		Bucket: bucket,
 		Prefix: options.Prefix,
-		More:   more,
+		More:   m,
 		Items:  objectsList,
+		Cursor: startAfterEnc,
 	}, nil
 }
 
-func (db *DB) objectsFromRawObjectList(ctx context.Context, items []RawObjectListItem, pi *encryption.PrefixInfo, startAfter string) (objectList []Object, err error) {
+func (db *DB) objectsFromRawObjectList(ctx context.Context, items []RawObjectListItem, pi *encryption.PrefixInfo) (objectList []Object, err error) {
 	objectList = make([]Object, 0, len(items))
 
 	for _, item := range items {

private/metaclient/types.go

@@ -165,6 +165,7 @@ const (
 type ListOptions struct {
 	Prefix                storj.Path
 	Cursor                storj.Path // Cursor is relative to Prefix, full path is Prefix + Cursor
+	CursorEnc             []byte
 	Delimiter             rune
 	Recursive             bool
 	Direction             ListDirection
@@ -182,7 +183,7 @@ func (opts ListOptions) NextPage(list ObjectList) ListOptions {
 
 	return ListOptions{
 		Prefix:                opts.Prefix,
-		Cursor:                list.Items[len(list.Items)-1].Path,
+		CursorEnc:             list.Cursor,
 		Delimiter:             opts.Delimiter,
 		Recursive:             opts.Recursive,
 		IncludeSystemMetadata: opts.IncludeSystemMetadata,
@@ -198,6 +199,7 @@ type ObjectList struct {
 	Bucket string
 	Prefix string
 	More   bool
+	Cursor []byte
 
 	// Items paths are relative to Prefix
 	// To get the full path use list.Prefix + list.Items[0].Path

private/metaclient/types_test.go

@@ -28,6 +28,7 @@ func TestListOptions(t *testing.T) {
 		Bucket: "hello",
 		Prefix: "alpha/",
 		More:   true,
+		Cursor: []byte("encrypted_path"),
 		Items: []metaclient.Object{
 			{Path: "alpha/xyz"},
 		},
@@ -36,7 +37,8 @@ func TestListOptions(t *testing.T) {
 	newopts := opts.NextPage(list)
 	require.Equal(t, metaclient.ListOptions{
 		Prefix:                "alpha/",
-		Cursor:                "alpha/xyz",
+		Cursor:                "",
+		CursorEnc:             []byte("encrypted_path"),
 		Delimiter:             '/',
 		Recursive:             true,
 		Direction:             metaclient.After,

testsuite/private/testuplink/objects_test.go

@@ -434,6 +434,68 @@ func TestListObjects(t *testing.T) {
 	})
 }
 
+func TestListObjects_PagingWithDiffPassphrase(t *testing.T) {
+	runTestWithPathCipher(t, 0, storj.EncAESGCM, func(t *testing.T, ctx context.Context, planet *testplanet.Planet, dbA *metaclient.DB, streams *streams.Store) {
+		bucket, err := dbA.CreateBucket(ctx, TestBucket)
+		require.NoError(t, err)
+
+		objectKeys := []string{
+			"a", "aa", "b", "bb", "c",
+			"a/xa", "a/xaa", "a/xb", "a/xbb", "a/xc",
+			"b/ya", "b/yaa", "b/yb", "b/ybb", "b/yc",
+		}
+
+		for _, key := range objectKeys {
+			upload(ctx, t, dbA, streams, bucket.Name, key, nil)
+		}
+
+		// Upload one object with different passphrase
+		encAccess := newTestEncStore(TestEncKey + "different")
+		encAccess.SetDefaultPathCipher(storj.EncAESGCM)
+
+		dbB, streams, cleanup, err := newMetainfoParts(planet, encAccess)
+		require.NoError(t, err)
+		defer func() {
+			err := cleanup()
+			assert.NoError(t, err)
+		}()
+		upload(ctx, t, dbB, streams, bucket.Name, "object_with_different_passphrase", nil)
+
+		for i, tt := range []struct {
+			options metaclient.ListOptions
+			more    bool
+			result  []string
+			db      *metaclient.DB
+		}{
+			{
+				options: optionsRecursive("", "", 1),
+				result:  []string{"object_with_different_passphrase"},
+				db:      dbB,
+				more:    true,
+			},
+		} {
+			errTag := fmt.Sprintf("%d. %+v", i, tt)
+
+			list, err := tt.db.ListObjects(ctx, bucket.Name, tt.options)
+
+			if assert.NoError(t, err, errTag) {
+				assert.Equal(t, tt.more, list.More, errTag)
+				if assert.Len(t, list.Items, len(tt.result), errTag) {
+					for i, item := range list.Items {
+						assert.Equal(t, tt.result[i], item.Path, errTag)
+						assert.Equal(t, TestBucket, item.Bucket.Name, errTag)
+						if item.IsPrefix {
+							assert.Equal(t, uint32(0), item.Version, errTag)
+						} else {
+							assert.Equal(t, uint32(1), item.Version, errTag)
+						}
+					}
+				}
+			}
+		}
+	})
+}
+
 func options(prefix, cursor string, limit int) metaclient.ListOptions {
 	return metaclient.ListOptions{
 		Prefix:    prefix,