feat: Add ListObjects gRPC method

- Evaluate minio errors. If the error is a permission related error, convert the error to a `Permission` proto and attach it to the gRPC status
- Update `dryRun` method to join all errors before returning response. This ensures all the permissions are validated before returning

plugin/service/storage/plugin.go

@@ -478,6 +478,179 @@ func (sp *StoragePlugin) GetObject(req *pb.GetObjectRequest, objServer pb.Storag
 	}
 }
 
+// listObjectsRecursive returns recursive list of all files and directories found under a prefix from MinIO.
+func listObjectsRecursive(ctx context.Context, cl *minio.Client, bucketName, searchPrefix string) ([]*pb.Object, error) {
+	var objects []*pb.Object
+	dirs := make(map[string]struct{})
+	seenKeys := make(map[string]struct{})
+
+	minioObjects := cl.ListObjects(ctx, bucketName, minio.ListObjectsOptions{
+		Prefix:    searchPrefix,
+		Recursive: true,
+	})
+
+	for objectName := range minioObjects {
+
+		// Bail out if MinIO reports an error for this object entry
+		if objectName.Err != nil {
+			return nil, objectName.Err
+		}
+		key := objectName.Key
+		isDirFile := strings.HasSuffix(key, "/")
+
+		// Skip the directory that is exactly the prefix itself (e.g., "foo/" when prefix == "foo/").
+		if isDirFile && key == searchPrefix {
+			continue
+		}
+
+		// Collect parent directories (normalize), but DO NOT mark them seen here.
+		// This ensures an explicit directory object (e.g., "folder/") will still be appended once.
+		dirPaths := buildRecursiveDirectoryPaths(key, searchPrefix)
+		for _, dirPath := range dirPaths {
+			k := ensureDir(dirPath)
+			dirs[k] = struct{}{}
+		}
+
+		// De-dupe only on the exact key string.
+		if _, exists := seenKeys[key]; exists {
+			continue
+		}
+		objects = append(objects, &pb.Object{Key: key, IsDir: isDirFile})
+		seenKeys[key] = struct{}{}
+	}
+
+	// Append synthesized directories, but only if an explicit object with the same key wasn't added.
+	for dirKey := range dirs {
+		if _, exists := seenKeys[dirKey]; exists {
+			continue
+		}
+		objects = append(objects, &pb.Object{Key: dirKey, IsDir: true})
+		seenKeys[dirKey] = struct{}{}
+	}
+	return objects, nil
+}
+
+// listObjectsNonRecursive returns only top-level files and directories within the prefix from MinIO.
+func listObjectsNonRecursive(ctx context.Context, cl *minio.Client, bucketName, searchPrefix string) ([]*pb.Object, error) {
+	var objects []*pb.Object
+	seenKeys := make(map[string]struct{})
+
+	minioObjects := cl.ListObjects(ctx, bucketName, minio.ListObjectsOptions{
+		Prefix:    searchPrefix,
+		Recursive: false,
+	})
+
+	for objectName := range minioObjects {
+		if objectName.Err != nil {
+			return nil, objectName.Err
+		}
+		key := objectName.Key
+		isDirFile := strings.HasSuffix(key, "/")
+
+		// Skip the prefix directory itself.
+		if isDirFile && key == searchPrefix {
+			continue
+		}
+
+		// Determine the top-level entry under the prefix.
+		relativeFullPath := strings.TrimSuffix(strings.TrimPrefix(key, searchPrefix), "/")
+		split := strings.Split(relativeFullPath, "/")
+		if len(split) == 0 {
+			continue
+		}
+		relativeParentPath := searchPrefix + split[0]
+
+		// If deeper segments or explicit dir file, normalize to a directory form.
+		if len(split) > 1 || isDirFile {
+			relativeParentPath = ensureDir(relativeParentPath)
+		}
+
+		if _, exists := seenKeys[relativeParentPath]; exists {
+			continue
+		}
+		objects = append(objects, &pb.Object{
+			Key:   relativeParentPath,
+			IsDir: strings.HasSuffix(relativeParentPath, "/"),
+		})
+		seenKeys[relativeParentPath] = struct{}{}
+	}
+	return objects, nil
+}
+
+// buildRecursiveDirectoryPaths collects all parent directory paths from an object key
+// that are within the given prefix. This is used to simulate directory structures in object storage.
+func buildRecursiveDirectoryPaths(objectKey, prefix string) []string {
+	var dirPaths []string
+	prefixDir := strings.TrimSuffix(prefix, "/")
+	pathParts := strings.Split(objectKey, "/")
+
+	// Build all possible parent directory paths from the split parts
+	for i := 1; i < len(pathParts); i++ {
+		dirPath := strings.Join(pathParts[:i], "/")
+
+		// Only include directories that are deeper than the search prefix
+		if len(dirPath) > len(prefixDir) && strings.HasPrefix(dirPath+"/", prefix) {
+			dirPaths = append(dirPaths, dirPath)
+		}
+	}
+	return dirPaths
+}
+
+func ensureDir(k string) string {
+	if strings.HasSuffix(k, "/") {
+		return k
+	}
+	return k + "/"
+}
+
+// ListObjects dispatches to recursive or non-recursive versions.
+func (sp *StoragePlugin) ListObjects(ctx context.Context, req *pb.ListObjectsRequest) (*pb.ListObjectsResponse, error) {
+	bucket := req.GetBucket()
+	if bucket == nil {
+		return nil, status.Error(codes.InvalidArgument, "no storage bucket information found")
+	}
+	if bucket.GetBucketName() == "" {
+		return nil, status.Error(codes.InvalidArgument, "storage bucket name is required")
+	}
+
+	// Build MinIO client (same pattern as HeadObject/GetObject/PutObject)
+	sa, err := getStorageAttributes(bucket.GetAttributes())
+	if err != nil {
+		return nil, err
+	}
+	sec, err := getStorageSecrets(bucket.GetSecrets())
+	if err != nil {
+		return nil, err
+	}
+	cl, err := minio.New(sa.EndpointUrl, &minio.Options{
+		Creds:  credentials.NewStaticV4(sec.AccessKeyId, sec.SecretAccessKey, ""),
+		Secure: sa.UseSSL,
+		Region: sa.Region,
+	})
+	if err != nil {
+		return nil, errorToStatus(codes.Unknown, "failed to create minio client", err, req).Err()
+	}
+
+	prefix := path.Join(bucket.GetBucketPrefix(), req.GetKeyPrefix())
+	// path.Join cleans the path, including removing a trailing slash, if it exists. Given that a slash is used to denote a folder,
+	// it is required here. So, if the original keyPrefix had a trailing slash, we need to add it back.
+	if strings.HasSuffix(req.GetKeyPrefix(), "/") {
+		prefix = prefix + "/"
+	}
+
+	var objs []*pb.Object
+	if req.GetRecursive() {
+		objs, err = listObjectsRecursive(ctx, cl, bucket.GetBucketName(), prefix)
+	} else {
+		objs, err = listObjectsNonRecursive(ctx, cl, bucket.GetBucketName(), prefix)
+	}
+	if err != nil {
+		return nil, errorToStatus(codes.Unknown, "failed to list objects", err, req).Err()
+	}
+
+	return &pb.ListObjectsResponse{Objects: objs}, nil
+}
+
 // PutObject is a hook that reads a file stored on local disk and stores it to
 // an external object store.
 func (sp *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest) (*pb.PutObjectResponse, error) {
@@ -873,6 +1046,8 @@ func errorToStatus(code codes.Code, msg string, err error, req any) *status.Stat
 		state.State.Read = permission
 	case pb.GetObjectRequest, *pb.GetObjectRequest:
 		state.State.Read = permission
+	case pb.ListObjectsRequest, *pb.ListObjectsRequest:
+		state.State.Read = permission
 	case pb.PutObjectRequest, *pb.PutObjectRequest:
 		state.State.Write = permission
 	case pb.DeleteObjectsRequest, *pb.DeleteObjectsRequest:

plugin/service/storage/plugin_test.go

@@ -14,6 +14,7 @@ import (
 	"io"
 	"os"
 	"path"
+	"strings"
 	"testing"
 	"time"
 
@@ -2895,6 +2896,356 @@ func TestPutObject(t *testing.T) {
 	}
 }
 
+// MockFile returns a file object (IsDir=false) for the given key.
+func MockFile(key string) *pb.Object {
+	return &pb.Object{Key: key, IsDir: false}
+}
+
+// MockDirectory returns a directory object (IsDir=true). Ensures trailing slash.
+func MockDirectory(key string) *pb.Object {
+	if !strings.HasSuffix(key, "/") {
+		key += "/"
+	}
+	return &pb.Object{Key: key, IsDir: true}
+}
+
+func TestListObjects(t *testing.T) {
+	ctx := context.Background()
+	server := internaltest.NewMinioServer(t)
+
+	bucketName := "test-bucket"
+	require.NoError(t, server.Client.MakeBucket(ctx, bucketName, minio.MakeBucketOptions{}))
+
+	seed := []string{
+		"test-file-1",
+		"test-file-2",
+		"folder1/file-1",
+		"folder1/subfolder/file-2",
+		"folder1/subfolder/nested/file-3",
+		"folder2/file-4",
+		"folder1/",
+		"folder1/subfolder/",
+		"folder1/subfolder2/",
+		"folder1/subfolder/nested/",
+		"folder2/",
+		"folder3/",
+	}
+	for _, k := range seed {
+		var (
+			r    = bytes.NewReader(nil)
+			size int64
+		)
+		if !strings.HasSuffix(k, "/") {
+			r = bytes.NewReader([]byte("x"))
+			size = 1
+		}
+		_, err := server.Client.PutObject(ctx, bucketName, k, r, size, minio.PutObjectOptions{})
+		require.NoError(t, err)
+	}
+
+	cases := []struct {
+		name            string
+		req             *pb.ListObjectsRequest
+		wantErrMsg      string
+		wantErrCode     codes.Code
+		expectedObjects []*pb.Object
+	}{
+		// Validation cases
+		{
+			name:        "nilRequest",
+			req:         nil,
+			wantErrMsg:  "no storage bucket information found",
+			wantErrCode: codes.InvalidArgument,
+		},
+		{
+			name:        "nilBucket",
+			req:         &pb.ListObjectsRequest{Bucket: nil},
+			wantErrMsg:  "no storage bucket information found",
+			wantErrCode: codes.InvalidArgument,
+		},
+		{
+			name:        "emptyBucket",
+			req:         &pb.ListObjectsRequest{Bucket: &storagebuckets.StorageBucket{}},
+			wantErrMsg:  "storage bucket name is required",
+			wantErrCode: codes.InvalidArgument,
+		},
+		{
+			name: "missing attributes",
+			req: &pb.ListObjectsRequest{
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: bucketName,
+					Secrets: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstAccessKeyId:     structpb.NewStringValue(server.ServiceAccountAccessKeyId),
+							ConstSecretAccessKey: structpb.NewStringValue(server.ServiceAccountSecretAccessKey),
+						},
+					},
+				},
+			},
+			wantErrMsg:  "empty attributes input",
+			wantErrCode: codes.InvalidArgument,
+		},
+		{
+			name: "badStorageAttributes",
+			req: &pb.ListObjectsRequest{
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: bucketName,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstEndpointUrl: structpb.NewStringValue("foo"), // invalid endpoint
+						},
+					},
+				},
+			},
+			wantErrMsg:  "attributes.endpoint_url.format: unknown protocol, should be http:// or https://",
+			wantErrCode: codes.InvalidArgument,
+		},
+		{
+			name: "badStorageSecrets",
+			req: &pb.ListObjectsRequest{
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: bucketName,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr),
+						},
+					},
+					Secrets: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstAccessKeyId: structpb.NewStringValue("foo"), // missing secret_access_key
+						},
+					},
+				},
+			},
+			wantErrMsg:  "secrets.secret_access_key: missing required value \"secret_access_key\"",
+			wantErrCode: codes.InvalidArgument,
+		},
+
+		{
+			name: "list all recursively",
+			req:  makeListRequest(bucketName, "", "", true, server, true),
+			expectedObjects: []*pb.Object{
+				MockFile("test-file-1"), MockFile("test-file-2"),
+				MockDirectory("folder1/"), MockFile("folder1/file-1"),
+				MockDirectory("folder1/subfolder/"), MockFile("folder1/subfolder/file-2"),
+				MockDirectory("folder1/subfolder/nested/"), MockFile("folder1/subfolder/nested/file-3"),
+				MockDirectory("folder1/subfolder2/"),
+				MockDirectory("folder2/"), MockFile("folder2/file-4"),
+				MockDirectory("folder3/"),
+			},
+		},
+		{
+			name: "list all non-recursively",
+			req:  makeListRequest(bucketName, "", "", false, server, true),
+			expectedObjects: []*pb.Object{
+				MockFile("test-file-1"), MockFile("test-file-2"),
+				MockDirectory("folder1/"), MockDirectory("folder2/"), MockDirectory("folder3/"),
+			},
+		},
+		{
+			name: "list objects with prefix folder1/ recursively",
+			req:  makeListRequest(bucketName, "", "folder1/", true, server, true),
+			expectedObjects: []*pb.Object{
+				MockFile("folder1/file-1"),
+				MockDirectory("folder1/subfolder/"), MockFile("folder1/subfolder/file-2"),
+				MockDirectory("folder1/subfolder/nested/"), MockFile("folder1/subfolder/nested/file-3"),
+				MockDirectory("folder1/subfolder2/"),
+			},
+		},
+		{
+			name: "list objects with prefix folder1/ non-recursive",
+			req:  makeListRequest(bucketName, "", "folder1/", false, server, true),
+			expectedObjects: []*pb.Object{
+				MockFile("folder1/file-1"),
+				MockDirectory("folder1/subfolder/"),
+				MockDirectory("folder1/subfolder2/"),
+			},
+		},
+		{
+			name: "list objects with prefix folder1/sub recursive",
+			req:  makeListRequest(bucketName, "", "folder1/sub", true, server, true),
+			expectedObjects: []*pb.Object{
+				MockDirectory("folder1/subfolder/"),
+				MockFile("folder1/subfolder/file-2"),
+				MockDirectory("folder1/subfolder/nested/"),
+				MockFile("folder1/subfolder/nested/file-3"),
+				MockDirectory("folder1/subfolder2/"),
+			},
+		},
+		{
+			name: "list objects with prefix folder1/sub non-recursive",
+			req:  makeListRequest(bucketName, "", "folder1/sub", false, server, true),
+			expectedObjects: []*pb.Object{
+				MockDirectory("folder1/subfolder/"),
+				MockDirectory("folder1/subfolder2/"),
+			},
+		},
+		{
+			name:            "list objects with prefix folder3/ recursively",
+			req:             makeListRequest(bucketName, "", "folder3/", true, server, true),
+			expectedObjects: []*pb.Object{},
+		},
+		{
+			name:            "list objects with prefix folder3/ non-recursively",
+			req:             makeListRequest(bucketName, "", "folder3/", false, server, true),
+			expectedObjects: []*pb.Object{},
+		},
+		{
+			name: "list objects with prefix folder1/subfolder2 recursively",
+			req:  makeListRequest(bucketName, "", "folder1/subfolder2", true, server, true),
+			expectedObjects: []*pb.Object{
+				MockDirectory("folder1/subfolder2/"),
+			},
+		},
+		{
+			name: "list objects with prefix folder1/subfolder2 non-recursively",
+			req:  makeListRequest(bucketName, "", "folder1/subfolder2", false, server, true),
+			expectedObjects: []*pb.Object{
+				MockDirectory("folder1/subfolder2/"),
+			},
+		},
+		{
+			name:            "list objects with prefix folder1/subfolder2/ recursively",
+			req:             makeListRequest(bucketName, "", "folder1/subfolder2/", true, server, true),
+			expectedObjects: []*pb.Object{},
+		},
+		{
+			name:            "list objects with prefix folder1/subfolder2/ non-recursively",
+			req:             makeListRequest(bucketName, "", "folder1/subfolder2/", false, server, true),
+			expectedObjects: []*pb.Object{},
+		},
+		{
+			name:            "list objects with non-existent prefix recursively",
+			req:             makeListRequest(bucketName, "", "nope/", true, server, true),
+			expectedObjects: []*pb.Object{},
+		},
+		{
+			name:            "list objects with non-existent prefix non-recursively",
+			req:             makeListRequest(bucketName, "", "nope/", false, server, true),
+			expectedObjects: []*pb.Object{},
+		},
+		{
+			name: "list all recursively with bucket prefix",
+			req:  makeListRequest(bucketName, "folder1/", "", true, server, true),
+			expectedObjects: []*pb.Object{
+				MockDirectory("folder1/"),
+				MockFile("folder1/file-1"),
+				MockDirectory("folder1/subfolder/"), MockFile("folder1/subfolder/file-2"),
+				MockDirectory("folder1/subfolder/nested/"), MockFile("folder1/subfolder/nested/file-3"),
+				MockDirectory("folder1/subfolder2/"),
+			},
+		},
+		{
+			name: "list all non-recursively with bucket prefix",
+			req:  makeListRequest(bucketName, "folder1/", "", false, server, true),
+			expectedObjects: []*pb.Object{
+				MockDirectory("folder1/"),
+			},
+		},
+		{
+			name: "list objects with bucket prefix and key prefix recursively",
+			req:  makeListRequest(bucketName, "folder1/", "sub", true, server, true),
+			expectedObjects: []*pb.Object{
+				MockDirectory("folder1/subfolder/"),
+				MockFile("folder1/subfolder/file-2"),
+				MockDirectory("folder1/subfolder/nested/"),
+				MockFile("folder1/subfolder/nested/file-3"),
+				MockDirectory("folder1/subfolder2/"),
+			},
+		},
+		{
+			name: "list objects with bucket prefix and key prefix non-recursively",
+			req:  makeListRequest(bucketName, "folder1/", "sub", false, server, true),
+			expectedObjects: []*pb.Object{
+				MockDirectory("folder1/subfolder/"),
+				MockDirectory("folder1/subfolder2/"),
+			},
+		},
+		{
+			name: "list objects with bucket prefix and trailing slash key prefix",
+			req:  makeListRequest(bucketName, "folder1/", "subfolder/", true, server, true),
+			expectedObjects: []*pb.Object{
+				MockFile("folder1/subfolder/file-2"),
+				MockDirectory("folder1/subfolder/nested/"),
+				MockFile("folder1/subfolder/nested/file-3"),
+			},
+		},
+		{
+			name: "list objects with bucket prefix pointing to empty area",
+			req:  makeListRequest(bucketName, "folder3/", "", true, server, true),
+			expectedObjects: []*pb.Object{
+				MockDirectory("folder3/"),
+			},
+		},
+		{
+			name:            "list objects with bucket prefix and non-existent key prefix",
+			req:             makeListRequest(bucketName, "folder1/", "nonexistent/", true, server, true),
+			expectedObjects: []*pb.Object{},
+		},
+
+		{
+			name:        "list without secrets non-recursive",
+			req:         makeListRequest(bucketName, "", "", false, server, false),
+			wantErrMsg:  "empty secrets input",
+			wantErrCode: codes.InvalidArgument,
+		},
+		{
+			name:        "list without secrets recursive",
+			req:         makeListRequest(bucketName, "", "", true, server, false),
+			wantErrMsg:  "empty secrets input",
+			wantErrCode: codes.InvalidArgument,
+		},
+	}
+
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			sp := new(StoragePlugin)
+			res, err := sp.ListObjects(ctx, tt.req)
+
+			if tt.wantErrMsg != "" {
+				require.Error(t, err)
+				require.Nil(t, res)
+				assert.ErrorContains(t, err, tt.wantErrMsg)
+				assert.Equal(t, tt.wantErrCode.String(), status.Code(err).String())
+				return
+			}
+
+			require.NoError(t, err)
+			require.NotNil(t, res)
+			assert.Equal(t, len(tt.expectedObjects), len(res.Objects), "object count mismatch")
+			assert.ElementsMatch(t, tt.expectedObjects, res.Objects, "object list mismatch")
+		})
+	}
+}
+
+// makeListRequest builds a ListObjectsRequest; when withSecrets=false, Secrets is omitted.
+// bucketPrefix can be empty string if no bucket prefix is needed.
+func makeListRequest(bucket, bucketPrefix, keyPrefix string, recursive bool, server *internaltest.MinioServer, withSecrets bool) *pb.ListObjectsRequest {
+	req := &pb.ListObjectsRequest{
+		KeyPrefix: keyPrefix,
+		Recursive: recursive,
+		Bucket: &storagebuckets.StorageBucket{
+			BucketName:   bucket,
+			BucketPrefix: bucketPrefix,
+			Attributes: &structpb.Struct{
+				Fields: map[string]*structpb.Value{
+					ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr),
+				},
+			},
+		},
+	}
+	if withSecrets {
+		req.Bucket.Secrets = &structpb.Struct{
+			Fields: map[string]*structpb.Value{
+				ConstAccessKeyId:     structpb.NewStringValue(server.ServiceAccountAccessKeyId),
+				ConstSecretAccessKey: structpb.NewStringValue(server.ServiceAccountSecretAccessKey),
+			},
+		}
+	}
+	return req
+}
+
 func TestDeleteObjects(t *testing.T) {
 	ctx := context.Background()
 	server := internaltest.NewMinioServer(t)