fix(s3api): make ListObjectsV1 namespaced and prevent marker-echo pagination loops (#8409)

* fix(s3api): make ListObjectsV1 namespaced and stop marker-echo pagination loops

* test(s3api): harden marker-echo coverage and align V1 encoding tag

* test(s3api): cover encoded marker matching and trim redundant setup

* refactor(s3api): tighten V1 list helper visibility and test mock docs

weed/s3api/s3api_object_handlers_list_test.go

@@ -64,6 +64,48 @@ func (c *testFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntr
 	return &testListEntriesStream{entries: entries}, nil
 }
 
+type markerEchoFilerClient struct {
+	filer_pb.SeaweedFilerClient
+	entriesByDir    map[string][]*filer_pb.Entry
+	returnFollowing bool
+}
+
+// markerEchoFilerClient intentionally ignores request Limit/InclusiveStartFrom
+// and simulates a backend that may echo StartFromFileName. entriesByDir controls
+// returned entries; returnFollowing controls whether ListEntries returns only the
+// echoed marker or the echoed marker plus following entries.
+func (c *markerEchoFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
+	entries := c.entriesByDir[in.Directory]
+	ensureEntryAttributes(entries)
+	if in.StartFromFileName == "" {
+		return &testListEntriesStream{entries: entries}, nil
+	}
+
+	for i, e := range entries {
+		if e.Name == in.StartFromFileName {
+			// Emulate buggy backend behavior: return marker again even when exclusive.
+			if c.returnFollowing {
+				echoAndFollowing := entries[i:]
+				return &testListEntriesStream{entries: echoAndFollowing}, nil
+			}
+			return &testListEntriesStream{entries: []*filer_pb.Entry{e}}, nil
+		}
+	}
+
+	return &testListEntriesStream{entries: nil}, nil
+}
+
+func ensureEntryAttributes(entries []*filer_pb.Entry) {
+	for _, entry := range entries {
+		if entry == nil {
+			continue
+		}
+		if entry.Attributes == nil {
+			entry.Attributes = &filer_pb.FuseAttributes{}
+		}
+	}
+}
+
 func TestListObjectsHandler(t *testing.T) {
 
 	// https://docs.aws.amazon.com/AmazonS3/latest/API/v2-RESTBucketGET.html
@@ -96,6 +138,20 @@ func TestListObjectsHandler(t *testing.T) {
 	}
 }
 
+func TestListObjectsV1NamespaceResponse(t *testing.T) {
+	response := ListBucketResult{
+		Name:        "test_container",
+		Prefix:      "",
+		Marker:      "",
+		NextMarker:  "",
+		MaxKeys:     1000,
+		IsTruncated: false,
+	}
+
+	encoded := string(s3err.EncodeXMLResponse(toListBucketResultV1(response)))
+	assert.Contains(t, encoded, `<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">`)
+}
+
 func Test_normalizePrefixMarker(t *testing.T) {
 	type args struct {
 		prefix string
@@ -253,6 +309,52 @@ func TestDoListFilerEntries_BucketRootPrefixSlashDelimiterSlash_ListsDirectories
 	assert.Contains(t, seen, "Veeam")
 }
 
+func TestDoListFilerEntries_ExclusiveStartSkipsMarkerEcho(t *testing.T) {
+	s3a := &S3ApiServer{}
+	client := &markerEchoFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket": {
+				{Name: "file.txt", Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "test.txt", Attributes: &filer_pb.FuseAttributes{}},
+			},
+		},
+	}
+
+	cursor := &ListingCursor{maxKeys: 1000}
+	var seen []string
+	nextMarker, err := s3a.doListFilerEntries(client, "/buckets/test-bucket", "", cursor, "test.txt", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+		seen = append(seen, entry.Name)
+	})
+
+	assert.NoError(t, err)
+	assert.Empty(t, seen, "marker entry should not be returned in exclusive mode")
+	assert.Equal(t, "", nextMarker, "next marker should be empty when only marker echo is returned")
+}
+
+func TestDoListFilerEntries_ExclusiveStartSkipsMarkerEchoWithSubsequentEntries(t *testing.T) {
+	s3a := &S3ApiServer{}
+	client := &markerEchoFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test-bucket": {
+				{Name: "file.txt", Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "test.txt", Attributes: &filer_pb.FuseAttributes{}},
+				{Name: "zebra.txt", Attributes: &filer_pb.FuseAttributes{}},
+			},
+		},
+		returnFollowing: true,
+	}
+
+	cursor := &ListingCursor{maxKeys: 1000}
+	var seen []string
+	nextMarker, err := s3a.doListFilerEntries(client, "/buckets/test-bucket", "", cursor, "test.txt", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
+		seen = append(seen, entry.Name)
+	})
+
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"zebra.txt"}, seen, "marker should be skipped while subsequent entries are returned")
+	assert.Equal(t, "zebra.txt", nextMarker)
+}
+
 func TestAllowUnorderedWithDelimiterValidation(t *testing.T) {
 	t.Run("should return error when allow-unordered=true and delimiter are both present", func(t *testing.T) {
 		// Create a request with both allow-unordered=true and delimiter
@@ -329,6 +431,38 @@ func TestAllowUnorderedWithDelimiterValidation(t *testing.T) {
 	})
 }
 
+func TestSanitizeV1MarkerEcho_NoProgressGuard(t *testing.T) {
+	response := ListBucketResult{
+		Marker:      "test.txt",
+		NextMarker:  "test.txt",
+		IsTruncated: true,
+		Contents: []ListEntry{
+			{Key: "test.txt"},
+		},
+	}
+
+	sanitizeV1MarkerEcho(&response, "test.txt", false)
+
+	assert.Empty(t, response.Contents)
+	assert.Equal(t, "", response.NextMarker)
+	assert.False(t, response.IsTruncated)
+
+	response2 := ListBucketResult{
+		Marker:      "test file.txt",
+		NextMarker:  "test%20file.txt",
+		IsTruncated: true,
+		Contents: []ListEntry{
+			{Key: "test%20file.txt"},
+		},
+	}
+
+	sanitizeV1MarkerEcho(&response2, "test file.txt", true)
+
+	assert.Empty(t, response2.Contents)
+	assert.Equal(t, "", response2.NextMarker)
+	assert.False(t, response2.IsTruncated)
+}
+
 // TestMaxKeysParameterValidation tests the validation of max-keys parameter
 func TestMaxKeysParameterValidation(t *testing.T) {
 	t.Run("valid max-keys values should work", func(t *testing.T) {