feat: storage plugin list objects (#6723)

* feat(proto): add storage plugin proto definitions for session recording recovery

Co-authored-by: Damian Debkowski <damian.debkowski@hashicorp.com>

Co-authored-by: Ryan Derr <ryan.derr@hashicorp.com>

* feat(gen): add generated storage plugin code for session recording recovery

Co-authored-by: Damian Debkowski <damian.debkowski@hashicorp.com>

Co-authored-by: Manjeet Singh <manjeet.singh@hashicorp.com>

* feat(plugin): add plugin loopback support for session recording recovery

Co-authored-by: Ryan Derr <ryan.derr@hashicorp.com>

Co-authored-by: aditya2548 <aditya25dec2000@gmail.com>

---------

Co-authored-by: Ryan Derr <ryan.derr@hashicorp.com>
Co-authored-by: Manjeet Singh <manjeet.singh@hashicorp.com>
Co-authored-by: aditya2548 <aditya25dec2000@gmail.com>

internal/plugin/loopback/client.go

@@ -91,6 +91,10 @@ func (tpc *WrappingPluginStorageClient) HeadObject(ctx context.Context, req *plg
 	return tpc.Server.HeadObject(ctx, req)
 }
 
+func (tpc *WrappingPluginStorageClient) ListObjects(ctx context.Context, req *plgpb.ListObjectsRequest, opts ...grpc.CallOption) (*plgpb.ListObjectsResponse, error) {
+	return tpc.Server.ListObjects(ctx, req)
+}
+
 func (tpc *WrappingPluginStorageClient) ValidatePermissions(ctx context.Context, req *plgpb.ValidatePermissionsRequest, opts ...grpc.CallOption) (*plgpb.ValidatePermissionsResponse, error) {
 	return tpc.Server.ValidatePermissions(ctx, req)
 }

internal/plugin/loopback/loopback.go

@@ -113,6 +113,7 @@ func NewLoopbackPlugin(opt ...TestOption) (*LoopbackPlugin, error) {
 	ret.GetObjectFn = ret.getObject
 	ret.PutObjectFn = ret.putObject
 	ret.DeleteObjectsFn = ret.deleteObjects
+	ret.ListObjectsFn = ret.listObjects
 	if len(opts.withMockBuckets) > 0 {
 		ret.buckets = opts.withMockBuckets
 	}

internal/plugin/loopback/options.go

@@ -4,6 +4,8 @@
 package loopback
 
 import (
+	"strings"
+
 	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
 	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
 	"google.golang.org/grpc/codes"
@@ -39,6 +41,9 @@ const (
 
 	// Will return an error for the DeleteObjects method
 	DeleteObjects
+
+	// Will return an error for the ListObjects method
+	ListObjects
 )
 
 const (
@@ -47,12 +52,47 @@ const (
 
 // PluginMockError is used to mock an error when interacting with an external object store.
 type PluginMockError struct {
-	BucketName                   string
-	BucketPrefix                 string
-	ObjectKey                    string
-	ErrMsg                       string
-	ErrCode                      codes.Code
-	ErrMethod                    Method
+	// BucketName is the name of the bucket to match on.
+	BucketName string
+
+	// BucketPrefix is the prefix of the bucket to match on.
+	BucketPrefix string
+
+	// ObjectKey is the object key to match on. This is optional and may be empty.
+	// If empty, the error will match any object key, as long as the ErrPath is not set.
+	// This is because the ErrPath is a less specific match than the ObjectKey.
+	// The ObjectKey is ignored when the ErrPath is set.
+	// The ObjectKey is also ignored for the following plugin methods: onCreateStorageBucket,
+	// onUpdateStorageBucket, onDeleteStorageBucket as these methods do not provide an object key.
+	ObjectKey string
+
+	// ErrPath is the object key path prefix to match on. This is optional and may be empty.
+	// If empty, the error will match any object key, as long as the ObjectKey is not set.
+	// The ErrPath is ignored when the ObjectKey is set.
+	// The ErrPath is also ignored for the following plugin methods: onCreateStorageBucket,
+	// onUpdateStorageBucket, onDeleteStorageBucket as these methods do not provide an object key.
+	// The ErrPath is used to match on a prefix of the object key. For example, if the ErrPath
+	// is set to "path/to/obj", it will match any object key that starts with "path/to/obj",
+	// such as "path/to/obj1", "path/to/obj2/subobj", etc.
+	// This allows for more flexible error mocking based on object key prefixes.
+	// The ErrPath should always end with a "/" if it is intended to match a directory prefix.
+	// For example, to match all objects under "path/to/dir/", the ErrPath should be set to "path/to/dir/".
+	// This ensures that it does not unintentionally match objects like "path/to/dir_obj".
+	ErrPath string
+
+	// ErrMsg is the error message to return.
+	ErrMsg string
+
+	// ErrCode is the gRPC error code to return.
+	ErrCode codes.Code
+
+	// ErrMethod is the plugin method to match on. This is optional and may be set to Any.
+	// If set to Any, the error will be returned for any plugin method that matches the bucket
+	// and object key criteria. If set to a specific method, the error will only be returned
+	// for that method if the bucket and object key criteria also match.
+	ErrMethod Method
+
+	// StorageBucketCredentialState is the credential state to return.
 	StorageBucketCredentialState *plgpb.StorageBucketCredentialState
 }
 
@@ -67,7 +107,8 @@ func (e PluginMockError) match(bucket *storagebuckets.StorageBucket, key string,
 	// the object key comparison is ignored when the given key is empty because the following
 	// plugin methods do not provide key values: onCreateStorageBucket, onUpdateStorageBucket,
 	// onDeleteStorageBucket
-	if key != "" && e.ObjectKey != key {
+	// if the mocked error object key is empty, it should match any key, as long as the ErrPath is not set.
+	if key != "" && e.ObjectKey != "" && e.ObjectKey != key {
 		return false
 	}
 	// if the mocked error bucket name does not match the request's bucket name, return false.
@@ -87,6 +128,10 @@ func (e PluginMockError) match(bucket *storagebuckets.StorageBucket, key string,
 	if e.ErrMethod != method {
 		return false
 	}
+	// if the mocked error path is not empty and the key does not have the mocked error path as a prefix, return false.
+	if e.ErrPath != "" && !strings.HasPrefix(key, e.ErrPath) {
+		return false
+	}
 	// all checks comparison checks passed, return true.
 	return true
 }

internal/plugin/loopback/options_test.go

@@ -0,0 +1,549 @@
+// Copyright IBM Corp. 2024, 2026
+// SPDX-License-Identifier: BUSL-1.1
+
+package loopback
+
+import (
+	"testing"
+
+	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
+	plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+)
+
+func TestPluginMockError_match(t *testing.T) {
+	t.Parallel()
+
+	testBucket := &storagebuckets.StorageBucket{
+		BucketName:   "test-bucket",
+		BucketPrefix: "test-prefix",
+	}
+
+	testBucketNoPrefix := &storagebuckets.StorageBucket{
+		BucketName: "test-bucket",
+	}
+
+	tests := []struct {
+		name      string
+		mockError PluginMockError
+		bucket    *storagebuckets.StorageBucket
+		key       string
+		method    Method
+		expected  bool
+	}{
+		{
+			name: "exact match with all fields",
+			mockError: PluginMockError{
+				BucketName:   "test-bucket",
+				BucketPrefix: "test-prefix",
+				ObjectKey:    "test-key",
+				ErrMethod:    PutObject,
+			},
+			bucket:   testBucket,
+			key:      "test-key",
+			method:   PutObject,
+			expected: true,
+		},
+		{
+			name: "match with Any method",
+			mockError: PluginMockError{
+				BucketName:   "test-bucket",
+				BucketPrefix: "test-prefix",
+				ObjectKey:    "test-key",
+				ErrMethod:    Any,
+			},
+			bucket:   testBucket,
+			key:      "test-key",
+			method:   GetObject,
+			expected: true,
+		},
+		{
+			name: "match with empty key (create/update/delete operations)",
+			mockError: PluginMockError{
+				BucketName:   "test-bucket",
+				BucketPrefix: "test-prefix",
+				ErrMethod:    OnCreateStorageBucket,
+			},
+			bucket:   testBucket,
+			key:      "", // Empty key for bucket operations
+			method:   OnCreateStorageBucket,
+			expected: true,
+		},
+		{
+			name: "match with path prefix",
+			mockError: PluginMockError{
+				BucketName:   "test-bucket",
+				BucketPrefix: "test-prefix",
+				ObjectKey:    "logs/session123/data.log", // Must match the key exactly
+				ErrPath:      "logs/",                    // Path prefix also matches
+				ErrMethod:    PutObject,
+			},
+			bucket:   testBucket,
+			key:      "logs/session123/data.log",
+			method:   PutObject,
+			expected: true,
+		},
+		{
+			name: "no match - different bucket name",
+			mockError: PluginMockError{
+				BucketName: "different-bucket",
+				ErrMethod:  PutObject,
+			},
+			bucket:   testBucket,
+			key:      "test-key",
+			method:   PutObject,
+			expected: false,
+		},
+		{
+			name: "no match - different object key",
+			mockError: PluginMockError{
+				BucketName: "test-bucket",
+				ObjectKey:  "different-key",
+				ErrMethod:  PutObject,
+			},
+			bucket:   testBucket,
+			key:      "test-key",
+			method:   PutObject,
+			expected: false,
+		},
+		{
+			name: "no match - different method",
+			mockError: PluginMockError{
+				BucketName: "test-bucket",
+				ObjectKey:  "test-key",
+				ErrMethod:  GetObject,
+			},
+			bucket:   testBucket,
+			key:      "test-key",
+			method:   PutObject,
+			expected: false,
+		},
+		{
+			name: "no match - different bucket prefix",
+			mockError: PluginMockError{
+				BucketName:   "test-bucket",
+				BucketPrefix: "different-prefix",
+				ErrMethod:    PutObject,
+			},
+			bucket:   testBucket,
+			key:      "test-key",
+			method:   PutObject,
+			expected: false,
+		},
+		{
+			name: "no match - path prefix doesn't match",
+			mockError: PluginMockError{
+				BucketName: "test-bucket",
+				ObjectKey:  "", // ObjectKey should be empty when using ErrPath
+				ErrPath:    "logs/",
+				ErrMethod:  PutObject,
+			},
+			bucket:   testBucketNoPrefix,
+			key:      "data/session123/info.json",
+			method:   PutObject,
+			expected: false,
+		},
+		{
+			name: "match - bucket with no prefix requirement",
+			mockError: PluginMockError{
+				BucketName: "test-bucket",
+				ObjectKey:  "test-key",
+				ErrMethod:  PutObject,
+			},
+			bucket:   testBucketNoPrefix,
+			key:      "test-key",
+			method:   PutObject,
+			expected: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := tt.mockError.match(tt.bucket, tt.key, tt.method)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}
+
+func TestPluginMockPutObjectResponse_match(t *testing.T) {
+	t.Parallel()
+
+	testBucket := &storagebuckets.StorageBucket{
+		BucketName:   "test-bucket",
+		BucketPrefix: "test-prefix",
+	}
+
+	testBucketNoPrefix := &storagebuckets.StorageBucket{
+		BucketName: "test-bucket",
+	}
+
+	tests := []struct {
+		name         string
+		mockResponse PluginMockPutObjectResponse
+		bucket       *storagebuckets.StorageBucket
+		key          string
+		expected     bool
+	}{
+		{
+			name: "exact match with all fields",
+			mockResponse: PluginMockPutObjectResponse{
+				BucketName:   "test-bucket",
+				BucketPrefix: "test-prefix",
+				ObjectKey:    "test-key",
+			},
+			bucket:   testBucket,
+			key:      "test-key",
+			expected: true,
+		},
+		{
+			name: "no match - different bucket name",
+			mockResponse: PluginMockPutObjectResponse{
+				BucketName: "different-bucket",
+				ObjectKey:  "test-key",
+			},
+			bucket:   testBucket,
+			key:      "test-key",
+			expected: false,
+		},
+		{
+			name: "no match - different object key",
+			mockResponse: PluginMockPutObjectResponse{
+				BucketName: "test-bucket",
+				ObjectKey:  "different-key",
+			},
+			bucket:   testBucket,
+			key:      "test-key",
+			expected: false,
+		},
+		{
+			name: "no match - different bucket prefix",
+			mockResponse: PluginMockPutObjectResponse{
+				BucketName:   "test-bucket",
+				BucketPrefix: "different-prefix",
+				ObjectKey:    "test-key",
+			},
+			bucket:   testBucket,
+			key:      "test-key",
+			expected: false,
+		},
+		{
+			name: "match - bucket with no prefix requirement",
+			mockResponse: PluginMockPutObjectResponse{
+				BucketName: "test-bucket",
+				ObjectKey:  "test-key",
+			},
+			bucket:   testBucketNoPrefix,
+			key:      "test-key",
+			expected: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := tt.mockResponse.match(tt.bucket, tt.key)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}
+
+func TestGetTestOpts(t *testing.T) {
+	t.Parallel()
+
+	t.Run("default options", func(t *testing.T) {
+		opts, err := getTestOpts()
+		require.NoError(t, err)
+
+		expected := TestOptions{
+			withChunkSize: 8,
+		}
+		assert.Equal(t, expected.withChunkSize, opts.withChunkSize)
+		assert.Nil(t, opts.withMockBuckets)
+		assert.Empty(t, opts.withMockError)
+		assert.Empty(t, opts.withMockPutObjectResponse)
+	})
+
+	t.Run("with mock buckets", func(t *testing.T) {
+		mockBuckets := map[BucketName]Bucket{
+			"test-bucket": make(Bucket),
+		}
+
+		opts, err := getTestOpts(WithMockBuckets(mockBuckets))
+		require.NoError(t, err)
+
+		assert.Equal(t, mockBuckets, opts.withMockBuckets)
+	})
+
+	t.Run("with mock error", func(t *testing.T) {
+		mockError := PluginMockError{
+			BucketName: "test-bucket",
+			ErrMsg:     "test error",
+			ErrCode:    codes.Internal,
+			ErrMethod:  PutObject,
+		}
+
+		opts, err := getTestOpts(WithMockError(mockError))
+		require.NoError(t, err)
+
+		require.Len(t, opts.withMockError, 1)
+		assert.Equal(t, mockError, opts.withMockError[0])
+	})
+
+	t.Run("with multiple mock errors", func(t *testing.T) {
+		mockError1 := PluginMockError{
+			BucketName: "test-bucket-1",
+			ErrMsg:     "test error 1",
+			ErrCode:    codes.Internal,
+			ErrMethod:  PutObject,
+		}
+		mockError2 := PluginMockError{
+			BucketName: "test-bucket-2",
+			ErrMsg:     "test error 2",
+			ErrCode:    codes.NotFound,
+			ErrMethod:  GetObject,
+		}
+
+		opts, err := getTestOpts(
+			WithMockError(mockError1),
+			WithMockError(mockError2),
+		)
+		require.NoError(t, err)
+
+		require.Len(t, opts.withMockError, 2)
+		assert.Equal(t, mockError1, opts.withMockError[0])
+		assert.Equal(t, mockError2, opts.withMockError[1])
+	})
+
+	t.Run("with mock put object response", func(t *testing.T) {
+		mockResponse := PluginMockPutObjectResponse{
+			BucketName: "test-bucket",
+			ObjectKey:  "test-key",
+			Response: &plgpb.PutObjectResponse{
+				ChecksumSha_256: []byte("test-checksum"),
+			},
+		}
+
+		opts, err := getTestOpts(WithMockPutObjectResponse(mockResponse))
+		require.NoError(t, err)
+
+		require.Len(t, opts.withMockPutObjectResponse, 1)
+		assert.Equal(t, mockResponse, opts.withMockPutObjectResponse[0])
+	})
+
+	t.Run("with custom chunk size", func(t *testing.T) {
+		customChunkSize := 16
+
+		opts, err := getTestOpts(WithChunkSize(customChunkSize))
+		require.NoError(t, err)
+
+		assert.Equal(t, customChunkSize, opts.withChunkSize)
+	})
+
+	t.Run("with all options combined", func(t *testing.T) {
+		mockBuckets := map[BucketName]Bucket{
+			"test-bucket": make(Bucket),
+		}
+		mockError := PluginMockError{
+			BucketName: "test-bucket",
+			ErrMsg:     "test error",
+			ErrCode:    codes.Internal,
+			ErrMethod:  PutObject,
+		}
+		mockResponse := PluginMockPutObjectResponse{
+			BucketName: "test-bucket",
+			ObjectKey:  "test-key",
+			Response: &plgpb.PutObjectResponse{
+				ChecksumSha_256: []byte("test-checksum"),
+			},
+		}
+		customChunkSize := 32
+
+		opts, err := getTestOpts(
+			WithMockBuckets(mockBuckets),
+			WithMockError(mockError),
+			WithMockPutObjectResponse(mockResponse),
+			WithChunkSize(customChunkSize),
+		)
+		require.NoError(t, err)
+
+		assert.Equal(t, mockBuckets, opts.withMockBuckets)
+		require.Len(t, opts.withMockError, 1)
+		assert.Equal(t, mockError, opts.withMockError[0])
+		require.Len(t, opts.withMockPutObjectResponse, 1)
+		assert.Equal(t, mockResponse, opts.withMockPutObjectResponse[0])
+		assert.Equal(t, customChunkSize, opts.withChunkSize)
+	})
+}
+
+func TestGetDefaultTestOptions(t *testing.T) {
+	t.Parallel()
+
+	opts := getDefaultTestOptions()
+
+	assert.Equal(t, 8, opts.withChunkSize)
+	assert.Nil(t, opts.withMockBuckets)
+	assert.Empty(t, opts.withMockError)
+	assert.Empty(t, opts.withMockPutObjectResponse)
+}
+
+func TestMethod_Constants(t *testing.T) {
+	t.Parallel()
+
+	// Test that Method constants have expected values
+	assert.Equal(t, Method(0), Any)
+	assert.Equal(t, Method(1), OnCreateStorageBucket)
+	assert.Equal(t, Method(2), OnUpdateStorageBucket)
+	assert.Equal(t, Method(3), OnDeleteStorageBucket)
+	assert.Equal(t, Method(4), ValidatePermissions)
+	assert.Equal(t, Method(5), HeadObject)
+	assert.Equal(t, Method(6), GetObject)
+	assert.Equal(t, Method(7), PutObject)
+	assert.Equal(t, Method(8), DeleteObjects)
+}
+
+func TestPluginMockError_ComplexScenarios(t *testing.T) {
+	t.Parallel()
+
+	t.Run("storage bucket credential state scenarios", func(t *testing.T) {
+		bucket := &storagebuckets.StorageBucket{
+			BucketName: "test-bucket",
+		}
+
+		credentialState := &plgpb.StorageBucketCredentialState{
+			State: &plgpb.Permissions{
+				Write: &plgpb.Permission{
+					State: plgpb.StateType_STATE_TYPE_ERROR,
+				},
+			},
+		}
+
+		mockError := PluginMockError{
+			BucketName:                   "test-bucket",
+			ErrMethod:                    ValidatePermissions,
+			StorageBucketCredentialState: credentialState,
+		}
+
+		result := mockError.match(bucket, "", ValidatePermissions)
+		assert.True(t, result)
+	})
+
+	t.Run("path-based error matching", func(t *testing.T) {
+		bucket := &storagebuckets.StorageBucket{
+			BucketName: "test-bucket",
+		}
+
+		tests := []struct {
+			name      string
+			errPath   string
+			objectKey string
+			key       string
+			expected  bool
+		}{
+			{
+				name:      "exact path match",
+				errPath:   "sessions/",
+				objectKey: "sessions/sr_123/data.log",
+				key:       "sessions/sr_123/data.log",
+				expected:  true,
+			},
+			{
+				name:      "nested path match",
+				errPath:   "sessions/sr_123/",
+				objectKey: "sessions/sr_123/connections/cr_456/data.log",
+				key:       "sessions/sr_123/connections/cr_456/data.log",
+				expected:  true,
+			},
+			{
+				name:      "no path match",
+				errPath:   "sessions/",
+				objectKey: "recordings/sr_123/data.log",
+				key:       "recordings/sr_123/data.log",
+				expected:  false,
+			},
+			{
+				name:      "path match but different object key",
+				errPath:   "sessions/",
+				objectKey: "sessions/file1.log",
+				key:       "sessions/file2.log",
+				expected:  false,
+			},
+		}
+
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				mockError := PluginMockError{
+					BucketName: "test-bucket",
+					ObjectKey:  tt.objectKey,
+					ErrPath:    tt.errPath,
+					ErrMethod:  PutObject,
+				}
+
+				result := mockError.match(bucket, tt.key, PutObject)
+				assert.Equal(t, tt.expected, result)
+			})
+		}
+	})
+
+	t.Run("path-based matching with empty key (bucket operations)", func(t *testing.T) {
+		bucket := &storagebuckets.StorageBucket{
+			BucketName: "test-bucket",
+		}
+
+		t.Run("empty path matches bucket operations", func(t *testing.T) {
+			mockError := PluginMockError{
+				BucketName: "test-bucket",
+				ObjectKey:  "",
+				ErrPath:    "", // Empty path works with empty key
+				ErrMethod:  OnCreateStorageBucket,
+			}
+
+			// Empty key for bucket operations (create/update/delete)
+			result := mockError.match(bucket, "", OnCreateStorageBucket)
+			assert.True(t, result)
+		})
+
+		t.Run("non-empty path does not match empty key", func(t *testing.T) {
+			mockError := PluginMockError{
+				BucketName: "test-bucket",
+				ObjectKey:  "",
+				ErrPath:    "some/path/", // Non-empty path won't match empty key
+				ErrMethod:  OnCreateStorageBucket,
+			}
+
+			// Empty key for bucket operations - should not match non-empty path
+			result := mockError.match(bucket, "", OnCreateStorageBucket)
+			assert.False(t, result)
+		})
+	})
+}
+
+func TestTestOptions_EdgeCases(t *testing.T) {
+	t.Parallel()
+
+	t.Run("zero chunk size", func(t *testing.T) {
+		opts, err := getTestOpts(WithChunkSize(0))
+		require.NoError(t, err)
+		assert.Equal(t, 0, opts.withChunkSize)
+	})
+
+	t.Run("negative chunk size", func(t *testing.T) {
+		opts, err := getTestOpts(WithChunkSize(-1))
+		require.NoError(t, err)
+		assert.Equal(t, -1, opts.withChunkSize)
+	})
+
+	t.Run("empty mock buckets map", func(t *testing.T) {
+		emptyBuckets := make(map[BucketName]Bucket)
+		opts, err := getTestOpts(WithMockBuckets(emptyBuckets))
+		require.NoError(t, err)
+		assert.NotNil(t, opts.withMockBuckets)
+		assert.Empty(t, opts.withMockBuckets)
+	})
+
+	t.Run("nil mock buckets map", func(t *testing.T) {
+		opts, err := getTestOpts(WithMockBuckets(nil))
+		require.NoError(t, err)
+		assert.Nil(t, opts.withMockBuckets)
+	})
+}

internal/plugin/loopback/storage.go

@@ -37,6 +37,7 @@ type TestPluginStorageServer struct {
 	GetObjectFn                  func(*plgpb.GetObjectRequest, plgpb.StoragePluginService_GetObjectServer) error
 	PutObjectFn                  func(context.Context, *plgpb.PutObjectRequest) (*plgpb.PutObjectResponse, error)
 	DeleteObjectsFn              func(context.Context, *plgpb.DeleteObjectsRequest) (*plgpb.DeleteObjectsResponse, error)
+	ListObjectsFn                func(context.Context, *plgpb.ListObjectsRequest) (*plgpb.ListObjectsResponse, error)
 	plgpb.UnimplementedStoragePluginServiceServer
 }
 
@@ -103,6 +104,13 @@ func (t TestPluginStorageServer) DeleteObjects(ctx context.Context, req *plgpb.D
 	return t.DeleteObjectsFn(ctx, req)
 }
 
+func (t TestPluginStorageServer) ListObjects(ctx context.Context, req *plgpb.ListObjectsRequest) (*plgpb.ListObjectsResponse, error) {
+	if t.ListObjectsFn == nil {
+		return t.UnimplementedStoragePluginServiceServer.ListObjects(ctx, req)
+	}
+	return t.ListObjectsFn(ctx, req)
+}
+
 type (
 	BucketName string
 	ObjectName string
@@ -157,9 +165,25 @@ func (l *LoopbackStorage) normalizeStorageBucketData(ctx context.Context, req *p
 // ResetNormalizations sets the number of times that NormalizeStorageBucketData
 // has been called to 0. Useful for unit tests.
 func (l *LoopbackStorage) ResetNormalizations() {
+	l.m.Lock()
+	defer l.m.Unlock()
 	l.normalizations = 0
 }
 
+// ResetMockErrors removes all mock errors.
+func (l *LoopbackStorage) ResetMockErrors() {
+	l.m.Lock()
+	defer l.m.Unlock()
+	l.errs = []PluginMockError{}
+}
+
+// SetMockErrors overrides the existing mock errors.
+func (l *LoopbackStorage) SetMockErrors(mockErrs ...PluginMockError) {
+	l.m.Lock()
+	defer l.m.Unlock()
+	l.errs = mockErrs
+}
+
 // GetNormalizations returns the number of times that NormalizeStorageBucketData
 // has been called via the loopback plugin. Useful for unit tests.
 func (l *LoopbackStorage) GetNormalizations() int {
@@ -538,6 +562,53 @@ func (l *LoopbackStorage) deleteObjects(ctx context.Context, req *plgpb.DeleteOb
 	}, nil
 }
 
+func (l *LoopbackStorage) listObjects(ctx context.Context, req *plgpb.ListObjectsRequest) (*plgpb.ListObjectsResponse, error) {
+	const op = "loopback.(LoopbackStorage).listObjects"
+
+	if req == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op)
+	}
+	if req.GetBucket() == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op)
+	}
+	if req.GetBucket().GetAttributes() == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "%s: missing bucket attributes", op)
+	}
+	l.m.Lock()
+	defer l.m.Unlock()
+	bucket, ok := l.buckets[BucketName(req.GetBucket().GetBucketName())]
+	if !ok {
+		return nil, status.Errorf(codes.NotFound, "%s: bucket not found", op)
+	}
+
+	// return an expected mock error if one was provided
+	for _, err := range l.errs {
+		if err.match(req.GetBucket(), req.GetKeyPrefix(), ListObjects) {
+			if err.StorageBucketCredentialState != nil {
+				return nil, createErrorWithBucketCredentialState(err.ErrCode, fmt.Sprintf("%s: %s", op, err.ErrMsg), err.StorageBucketCredentialState)
+			}
+			return nil, status.Errorf(err.ErrCode, "%s: %s", op, err.ErrMsg)
+		}
+	}
+
+	searchPrefix := path.Join(req.GetBucket().GetBucketPrefix(), req.GetKeyPrefix())
+	// path.Join strips trailing slashes, but we need to preserve them for directory prefixes
+	// to ensure proper filtering behavior in listObjects functions
+	if strings.HasSuffix(req.GetKeyPrefix(), "/") || (req.GetKeyPrefix() == "" && strings.HasSuffix(req.GetBucket().GetBucketPrefix(), "/")) {
+		searchPrefix = searchPrefix + "/"
+	}
+
+	var objects []*plgpb.Object
+	if req.GetRecursive() {
+		objects = listObjectsRecursive(bucket, searchPrefix)
+	} else {
+		objects = listObjectsNonRecursive(bucket, searchPrefix)
+	}
+	return &plgpb.ListObjectsResponse{
+		Objects: objects,
+	}, nil
+}
+
 // CloneBucket returns a clone of the bucket.
 // returns nil when the bucket is not found.
 func (l *LoopbackStorage) CloneBucket(name string) Bucket {
@@ -612,3 +683,146 @@ func createErrorWithBucketCredentialState(errCode codes.Code, msg string, sbcSta
 	}
 	return st.Err()
 }
+
+// listObjectsRecursive returns recursive list of all files and directories found under a prefix from a provided bucket
+func listObjectsRecursive(bucket Bucket, searchPrefix string) []*plgpb.Object {
+	var objects []*plgpb.Object
+	dirs := make(map[string]struct{})
+	seenKeys := make(map[string]struct{})
+
+	for objectName := range bucket {
+		key := string(objectName)
+		// Skip any objects that do not match the search prefix
+		if !strings.HasPrefix(key, searchPrefix) {
+			continue
+		}
+
+		// Skip any objects that have already been seen
+		if _, exists := seenKeys[key]; exists {
+			continue
+		}
+
+		isDirFile := strings.HasSuffix(key, "/")
+		// If the object is a directory and matches the search prefix, skip it
+		// This can happen if the prefix itself is a directory
+		// e.g. prefix = "foo/", object key = "foo/"
+		// In this case we don't want to return the directory itself as an object
+		if isDirFile && key == searchPrefix {
+			continue
+		}
+
+		dirPaths := buildRecursiveDirectoryPaths(key, searchPrefix)
+		for _, dirPath := range dirPaths {
+			dirs[dirPath] = struct{}{}
+			seenKeys[dirPath] = struct{}{}
+		}
+
+		// Only add the object if it hasn't already been added as a directory
+		// This can happen if the object key ends with a '/' and is added as a directory
+		// in the previous step
+		if _, exists := seenKeys[key]; !exists {
+			objects = append(objects, &plgpb.Object{
+				Key:   key,
+				IsDir: isDirFile,
+			})
+		}
+		seenKeys[key] = struct{}{}
+	}
+
+	for dirKey := range dirs {
+		objects = append(objects, &plgpb.Object{
+			Key:   dirKey,
+			IsDir: true,
+		})
+	}
+
+	return objects
+}
+
+// listObjectsNonRecursive returns only top-level files and directories within the prefix
+func listObjectsNonRecursive(bucket Bucket, searchPrefix string) []*plgpb.Object {
+	var objects []*plgpb.Object
+	seenKeys := make(map[string]struct{})
+
+	for objectName := range bucket {
+		key := string(objectName)
+		// Skip any objects that do not match the search prefix
+		if !strings.HasPrefix(key, searchPrefix) {
+			continue
+		}
+
+		// Skip any objects that have already been seen
+		if _, exists := seenKeys[key]; exists {
+			continue
+		}
+
+		isDirFile := strings.HasSuffix(key, "/")
+		// If the object is a directory and matches the search prefix, skip it
+		// This can happen if the prefix itself is a directory
+		// e.g. prefix = "foo/", object key = "foo/"
+		// In this case we don't want to return the directory itself as an object
+		if isDirFile && key == searchPrefix {
+			continue
+		}
+
+		// Trim the prefix and and trailing '/' after which we split the path
+		// to determine if this object is in a subdirectory or a file at the root prefix
+		relativeFullPath := strings.TrimSuffix(strings.TrimPrefix(key, searchPrefix), "/")
+		split := strings.Split(relativeFullPath, "/")
+		// Adding this to guardrail against the case that the stdlib updates the behavior
+		// of strings.Split() causing it to not always return at least a 1 length slice
+		if len(split) == 0 {
+			continue
+		}
+		relativeParentPath := searchPrefix + split[0]
+
+		// If the original object key ends with a '/' we treat it as a directory
+		// even if it is at the root of the prefix.
+		// If the split contains more than one entry, it is also a directory as it has sub-paths
+		if len(split) > 1 || isDirFile {
+			relativeParentPath += "/"
+		}
+
+		// Skip any constructed parent paths that have already been seen
+		if _, exists := seenKeys[relativeParentPath]; exists {
+			continue
+		}
+
+		// If the split contains more than one entry, it is a directory as it has sub-paths
+		// If the original object key ends with a '/' we treat it as a directory
+		// even if it is at the root of the prefix.
+		if len(split) > 1 {
+			objects = append(objects, &plgpb.Object{
+				Key:   relativeParentPath,
+				IsDir: true,
+			})
+		} else {
+			objects = append(objects, &plgpb.Object{
+				Key:   relativeParentPath,
+				IsDir: strings.HasSuffix(relativeParentPath, "/"),
+			})
+		}
+		seenKeys[relativeParentPath] = struct{}{}
+	}
+
+	return objects
+}
+
+// buildDirectoryPaths 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 and add a trailing slash to denote a directory
+		if len(dirPath) > len(prefixDir) && strings.HasPrefix(dirPath+"/", prefix) {
+			dirPaths = append(dirPaths, fmt.Sprintf("%s/", dirPath))
+		}
+	}
+	return dirPaths
+}

internal/plugin/loopback/storage_test.go

@@ -1618,6 +1618,611 @@ func TestLoopbackDeleteObjects(t *testing.T) {
 	}
 }
 
+func TestLoopbackListObjects(t *testing.T) {
+	require, assert := tr.New(t), ta.New(t)
+
+	mockStorageMapData := map[BucketName]Bucket{
+		"aws_s3_mock": {
+			"test-file-1": MockObject([]Chunk{
+				[]byte("content of test-file-1"),
+			}),
+			"folder1/": MockObject(nil),
+			"test-file-2": MockObject([]Chunk{
+				[]byte("content of test-file-2"),
+			}),
+			"folder1/file-1": MockObject([]Chunk{
+				[]byte("content of folder1/file-1"),
+			}),
+			"folder1/subfolder/": MockObject(nil),
+			"folder1/subfolder/file-2": MockObject([]Chunk{
+				[]byte("content of folder1/subfolder/file-2"),
+			}),
+			"folder1/subfolder/nested/file-3": MockObject([]Chunk{
+				[]byte("content of folder1/subfolder/nested/file-3"),
+			}),
+			"folder1/subfolder2/": MockObject(nil),
+			"folder2/file-4": MockObject([]Chunk{
+				[]byte("content of folder2/file-4"),
+			}),
+			"folder3/": MockObject(nil),
+		},
+		"aws_s3_err": {
+			"test-file": MockObject([]Chunk{
+				[]byte("test content"),
+			}),
+		},
+	}
+
+	plg, err := NewLoopbackPlugin(
+		WithMockBuckets(mockStorageMapData),
+		WithMockError(PluginMockError{
+			BucketName: "aws_s3_err",
+			ErrMsg:     "invalid credentials",
+			ErrCode:    codes.PermissionDenied,
+		}),
+	)
+	assert.NoError(err)
+
+	client := NewWrappingPluginStorageClient(plg)
+	secrets := &structpb.Struct{
+		Fields: map[string]*structpb.Value{
+			"AWS_ACCESS_KEY_ID":     structpb.NewStringValue("access_key_id"),
+			"AWS_SECRET_ACCESS_KEY": structpb.NewStringValue("secret_access_key"),
+		},
+	}
+
+	tests := []struct {
+		name            string
+		request         *plgpb.ListObjectsRequest
+		expectedErr     codes.Code
+		expectedObjects []*plgpb.Object
+	}{
+		{
+			name:        "missing request",
+			expectedErr: codes.InvalidArgument,
+		},
+		{
+			name:        "missing storage bucket",
+			request:     &plgpb.ListObjectsRequest{},
+			expectedErr: codes.InvalidArgument,
+		},
+		{
+			name: "missing attributes",
+			request: &plgpb.ListObjectsRequest{
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+				},
+			},
+			expectedErr: codes.InvalidArgument,
+		},
+		{
+			name: "bucket not found",
+			request: &plgpb.ListObjectsRequest{
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "invalid_bucket",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedErr: codes.NotFound,
+		},
+		{
+			name: "mocked error",
+			request: &plgpb.ListObjectsRequest{
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_err",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedErr: codes.PermissionDenied,
+		},
+		{
+			name: "list all objects without prefix recursively",
+			request: &plgpb.ListObjectsRequest{
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "test-file-1", IsDir: false},
+				{Key: "test-file-2", IsDir: false},
+				{Key: "folder1/", IsDir: true},
+				{Key: "folder1/file-1", IsDir: false},
+				{Key: "folder1/subfolder/", IsDir: true},
+				{Key: "folder1/subfolder2/", IsDir: true},
+				{Key: "folder1/subfolder/file-2", IsDir: false},
+				{Key: "folder1/subfolder/nested/", IsDir: true},
+				{Key: "folder1/subfolder/nested/file-3", IsDir: false},
+				{Key: "folder2/", IsDir: true},
+				{Key: "folder2/file-4", IsDir: false},
+				{Key: "folder3/", IsDir: true},
+			},
+		},
+		{
+			name: "list all objects without prefix non-recursively",
+			request: &plgpb.ListObjectsRequest{
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "test-file-1", IsDir: false},
+				{Key: "test-file-2", IsDir: false},
+				{Key: "folder1/", IsDir: true},
+				{Key: "folder2/", IsDir: true},
+				{Key: "folder3/", IsDir: true},
+			},
+		},
+		{
+			name: "list objects with prefix folder1/ recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder1/",
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/file-1", IsDir: false},
+				{Key: "folder1/subfolder/", IsDir: true},
+				{Key: "folder1/subfolder2/", IsDir: true},
+				{Key: "folder1/subfolder/file-2", IsDir: false},
+				{Key: "folder1/subfolder/nested/", IsDir: true},
+				{Key: "folder1/subfolder/nested/file-3", IsDir: false},
+			},
+		},
+		{
+			name: "list objects with prefix folder1/ non-recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder1/",
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/file-1", IsDir: false},
+				{Key: "folder1/subfolder/", IsDir: true},
+				{Key: "folder1/subfolder2/", IsDir: true},
+			},
+		},
+		{
+			name: "list objects with prefix folder1/sub recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder1/sub",
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/subfolder/", IsDir: true},
+				{Key: "folder1/subfolder2/", IsDir: true},
+				{Key: "folder1/subfolder/file-2", IsDir: false},
+				{Key: "folder1/subfolder/nested/", IsDir: true},
+				{Key: "folder1/subfolder/nested/file-3", IsDir: false},
+			},
+		},
+		{
+			name: "list objects with prefix folder1/sub non-recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder1/sub",
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/subfolder/", IsDir: true},
+				{Key: "folder1/subfolder2/", IsDir: true},
+			},
+		},
+		{
+			name: "list objects with prefix folder3/ recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder3/",
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{},
+		},
+		{
+			name: "list objects with prefix folder3/ non-recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder3/",
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{},
+		},
+		{
+			name: "list objects with prefix folder1/subfolder2 recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder1/subfolder2",
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/subfolder2/", IsDir: true},
+			},
+		},
+		{
+			name: "list objects with prefix folder1/subfolder2 non-recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder1/subfolder2",
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/subfolder2/", IsDir: true},
+			},
+		},
+		{
+			name: "list objects with prefix folder1/subfolder2/ recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder1/subfolder2/",
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{},
+		},
+		{
+			name: "list objects with prefix folder1/subfolder2/ non-recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder1/subfolder2/",
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{},
+		},
+		{
+			name: "list objects with non-existent prefix recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "nonexistent/",
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{},
+		},
+		{
+			name: "list objects with non-existent prefix non-recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "nonexistent/",
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Secrets:    secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{},
+		},
+		{
+			name: "list objects without secrets non-recursively",
+			request: &plgpb.ListObjectsRequest{
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "test-file-1", IsDir: false},
+				{Key: "test-file-2", IsDir: false},
+				{Key: "folder1/", IsDir: true},
+				{Key: "folder2/", IsDir: true},
+				{Key: "folder3/", IsDir: true},
+			},
+		},
+		{
+			name: "list objects without secrets recursively",
+			request: &plgpb.ListObjectsRequest{
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "aws_s3_mock",
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "test-file-1", IsDir: false},
+				{Key: "test-file-2", IsDir: false},
+				{Key: "folder1/", IsDir: true},
+				{Key: "folder1/file-1", IsDir: false},
+				{Key: "folder1/subfolder/", IsDir: true},
+				{Key: "folder1/subfolder2/", IsDir: true},
+				{Key: "folder1/subfolder/file-2", IsDir: false},
+				{Key: "folder1/subfolder/nested/", IsDir: true},
+				{Key: "folder1/subfolder/nested/file-3", IsDir: false},
+				{Key: "folder2/", IsDir: true},
+				{Key: "folder2/file-4", IsDir: false},
+				{Key: "folder3/", IsDir: true},
+			},
+		},
+		{
+			name: "list objects with bucket prefix and no key prefix recursively",
+			request: &plgpb.ListObjectsRequest{
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName:   "aws_s3_mock",
+					BucketPrefix: "folder1/",
+					Secrets:      secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/file-1", IsDir: false},
+				{Key: "folder1/subfolder/", IsDir: true},
+				{Key: "folder1/subfolder2/", IsDir: true},
+				{Key: "folder1/subfolder/file-2", IsDir: false},
+				{Key: "folder1/subfolder/nested/", IsDir: true},
+				{Key: "folder1/subfolder/nested/file-3", IsDir: false},
+			},
+		},
+		{
+			name: "list objects with bucket prefix and no key prefix non-recursively",
+			request: &plgpb.ListObjectsRequest{
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName:   "aws_s3_mock",
+					BucketPrefix: "folder1/",
+					Secrets:      secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/file-1", IsDir: false},
+				{Key: "folder1/subfolder/", IsDir: true},
+				{Key: "folder1/subfolder2/", IsDir: true},
+			},
+		},
+		{
+			name: "list objects with bucket prefix and key prefix recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "subfolder/",
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName:   "aws_s3_mock",
+					BucketPrefix: "folder1/",
+					Secrets:      secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/subfolder/file-2", IsDir: false},
+				{Key: "folder1/subfolder/nested/", IsDir: true},
+				{Key: "folder1/subfolder/nested/file-3", IsDir: false},
+			},
+		},
+		{
+			name: "list objects with bucket prefix and key prefix non-recursively",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "subfolder/",
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName:   "aws_s3_mock",
+					BucketPrefix: "folder1/",
+					Secrets:      secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/subfolder/file-2", IsDir: false},
+				{Key: "folder1/subfolder/nested/", IsDir: true},
+			},
+		},
+		{
+			name: "list objects with bucket prefix no trailing slash and key prefix with trailing slash",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "subfolder/",
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName:   "aws_s3_mock",
+					BucketPrefix: "folder1",
+					Secrets:      secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/subfolder/file-2", IsDir: false},
+				{Key: "folder1/subfolder/nested/", IsDir: true},
+				{Key: "folder1/subfolder/nested/file-3", IsDir: false},
+			},
+		},
+		{
+			name: "list objects with bucket prefix empty and key prefix folder1/",
+			request: &plgpb.ListObjectsRequest{
+				KeyPrefix: "folder1/",
+				Recursive: false,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName:   "aws_s3_mock",
+					BucketPrefix: "",
+					Secrets:      secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{
+				{Key: "folder1/file-1", IsDir: false},
+				{Key: "folder1/subfolder/", IsDir: true},
+				{Key: "folder1/subfolder2/", IsDir: true},
+			},
+		},
+		{
+			name: "list objects with bucket prefix filter matching nothing",
+			request: &plgpb.ListObjectsRequest{
+				Recursive: true,
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName:   "aws_s3_mock",
+					BucketPrefix: "nonexistent/",
+					Secrets:      secrets,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							"endpoint": structpb.NewStringValue("0.0.0.0"),
+						},
+					},
+				},
+			},
+			expectedObjects: []*plgpb.Object{},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			resp, err := client.ListObjects(context.Background(), tt.request)
+			if tt.expectedErr != codes.OK {
+				assert.Error(err)
+				assert.Nil(resp)
+				s, ok := status.FromError(err)
+				require.True(ok, "invalid error type")
+				assert.Equal(tt.expectedErr, s.Code())
+				return
+			}
+			assert.NoError(err)
+			require.NotNil(resp)
+
+			require.Equal(len(tt.expectedObjects), len(resp.Objects), "object count mismatch")
+			assert.ElementsMatch(tt.expectedObjects, resp.Objects, "object list mismatch")
+		})
+	}
+}
+
 func TestLoopbackStoragePlugin(t *testing.T) {
 	td := t.TempDir()