feat: Adds Ownder data in ListObjects(V2) result.

Closes #819

ListObjects returns object owner data in each object entity in the result, while ListObjectsV2 has fetch-owner query param, which indicates if the objects owner data should be fetched.
Adds these changes in the gateway to add `Owner` data in `ListObjects` and `ListObjectsV2` result. In aws the objects can be owned by different users in the same bucket. In the gateway all the objects are owned by the bucket owner.

backend/azure/azure.go

@@ -584,6 +584,18 @@ func (az *Azure) ListObjects(ctx context.Context, input *s3.ListObjectsInput) (s
 		maxKeys = *input.MaxKeys
 	}
 
+	// Retrieve the bucket acl to get the bucket owner
+	// All the objects in the bucket are owner by the bucket owner
+	aclBytes, err := az.getContainerMetaData(ctx, *input.Bucket, string(keyAclCapital))
+	if err != nil {
+		return s3response.ListObjectsResult{}, azureErrToS3Err(err)
+	}
+
+	var acl auth.ACL
+	if err := json.Unmarshal(aclBytes, &acl); err != nil {
+		return s3response.ListObjectsResult{}, fmt.Errorf("unmarshal acl: %w", err)
+	}
+
 Pager:
 	for pager.More() {
 		resp, err := pager.NextPage(ctx)
@@ -602,6 +614,9 @@ Pager:
 				LastModified: v.Properties.LastModified,
 				Size:         v.Properties.ContentLength,
 				StorageClass: types.ObjectStorageClassStandard,
+				Owner: &types.Owner{
+					ID: &acl.Owner,
+				},
 			})
 		}
 		for _, v := range resp.Segment.BlobPrefixes {
@@ -661,10 +676,28 @@ func (az *Azure) ListObjectsV2(ctx context.Context, input *s3.ListObjectsV2Input
 	var nextMarker *string
 	var isTruncated bool
 	var maxKeys int32 = math.MaxInt32
+	var fetchOwner bool
 
 	if input.MaxKeys != nil {
 		maxKeys = *input.MaxKeys
 	}
+	if input.FetchOwner != nil {
+		fetchOwner = *input.FetchOwner
+	}
+
+	// Retrieve the bucket acl to get the bucket owner, if "fetchOwner" is true
+	// All the objects in the bucket are owner by the bucket owner
+	var acl auth.ACL
+	if fetchOwner {
+		aclBytes, err := az.getContainerMetaData(ctx, *input.Bucket, string(keyAclCapital))
+		if err != nil {
+			return s3response.ListObjectsV2Result{}, azureErrToS3Err(err)
+		}
+
+		if err := json.Unmarshal(aclBytes, &acl); err != nil {
+			return s3response.ListObjectsV2Result{}, fmt.Errorf("unmarshal acl: %w", err)
+		}
+	}
 
 Pager:
 	for pager.More() {
@@ -678,13 +711,20 @@ Pager:
 				isTruncated = true
 				break Pager
 			}
-			objects = append(objects, s3response.Object{
+
+			obj := s3response.Object{
 				ETag:         backend.GetPtrFromString(fmt.Sprintf("%q", *v.Properties.ETag)),
 				Key:          v.Name,
 				LastModified: v.Properties.LastModified,
 				Size:         v.Properties.ContentLength,
 				StorageClass: types.ObjectStorageClassStandard,
-			})
+			}
+			if fetchOwner {
+				obj.Owner = &types.Owner{
+					ID: &acl.Owner,
+				}
+			}
+			objects = append(objects, obj)
 		}
 		for _, v := range resp.Segment.BlobPrefixes {
 			if *v.Name <= marker {

backend/posix/posix.go

@@ -4143,7 +4143,7 @@ func (p *Posix) ListObjects(ctx context.Context, input *s3.ListObjectsInput) (s3
 
 	fileSystem := os.DirFS(bucket)
 	results, err := backend.Walk(ctx, fileSystem, prefix, delim, marker, maxkeys,
-		p.fileToObj(bucket), []string{metaTmpDir})
+		p.fileToObj(bucket, true), []string{metaTmpDir})
 	if err != nil {
 		return s3response.ListObjectsResult{}, fmt.Errorf("walk %v: %w", bucket, err)
 	}
@@ -4161,8 +4161,25 @@ func (p *Posix) ListObjects(ctx context.Context, input *s3.ListObjectsInput) (s3
 	}, nil
 }
 
-func (p *Posix) fileToObj(bucket string) backend.GetObjFunc {
+func (p *Posix) fileToObj(bucket string, fetchOwner bool) backend.GetObjFunc {
 	return func(path string, d fs.DirEntry) (s3response.Object, error) {
+		var owner *types.Owner
+		// Retreive the object owner data from bucket ACL, if fetchOwner is true
+		// All the objects in the bucket are owned by the bucket owner
+		if fetchOwner {
+			aclJSON, err := p.meta.RetrieveAttribute(nil, bucket, "", aclkey)
+			if err != nil {
+				return s3response.Object{}, fmt.Errorf("get bucket acl: %w", err)
+			}
+			var acl auth.ACL
+			if err := json.Unmarshal(aclJSON, &acl); err != nil {
+				return s3response.Object{}, fmt.Errorf("unmarshal acl: %w", err)
+			}
+
+			owner = &types.Owner{
+				ID: &acl.Owner,
+			}
+		}
 		if d.IsDir() {
 			// directory object only happens if directory empty
 			// check to see if this is a directory object by checking etag
@@ -4192,6 +4209,7 @@ func (p *Posix) fileToObj(bucket string) backend.GetObjFunc {
 				LastModified: &mtime,
 				Size:         &size,
 				StorageClass: types.ObjectStorageClassStandard,
+				Owner:        owner,
 			}, nil
 		}
 
@@ -4239,6 +4257,7 @@ func (p *Posix) fileToObj(bucket string) backend.GetObjFunc {
 			StorageClass:      types.ObjectStorageClassStandard,
 			ChecksumAlgorithm: []types.ChecksumAlgorithm{checksums.Algorithm},
 			ChecksumType:      checksums.Type,
+			Owner:             owner,
 		}, nil
 	}
 }
@@ -4272,6 +4291,10 @@ func (p *Posix) ListObjectsV2(ctx context.Context, input *s3.ListObjectsV2Input)
 	if input.MaxKeys != nil {
 		maxkeys = *input.MaxKeys
 	}
+	var fetchOwner bool
+	if input.FetchOwner != nil {
+		fetchOwner = *input.FetchOwner
+	}
 
 	_, err := os.Stat(bucket)
 	if errors.Is(err, fs.ErrNotExist) {
@@ -4283,7 +4306,7 @@ func (p *Posix) ListObjectsV2(ctx context.Context, input *s3.ListObjectsV2Input)
 
 	fileSystem := os.DirFS(bucket)
 	results, err := backend.Walk(ctx, fileSystem, prefix, delim, marker, maxkeys,
-		p.fileToObj(bucket), []string{metaTmpDir})
+		p.fileToObj(bucket, fetchOwner), []string{metaTmpDir})
 	if err != nil {
 		return s3response.ListObjectsV2Result{}, fmt.Errorf("walk %v: %w", bucket, err)
 	}

s3api/controllers/base.go

@@ -1119,6 +1119,8 @@ func (c S3ApiController) ListActions(ctx *fiber.Ctx) error {
 					BucketOwner: parsedAcl.Owner,
 				})
 		}
+
+		fetchOwner := strings.EqualFold(ctx.Query("fetch-owner"), "true")
 		res, err := c.be.ListObjectsV2(ctx.Context(),
 			&s3.ListObjectsV2Input{
 				Bucket:            &bucket,
@@ -1127,6 +1129,7 @@ func (c S3ApiController) ListActions(ctx *fiber.Ctx) error {
 				Delimiter:         &delimiter,
 				MaxKeys:           &maxkeys,
 				StartAfter:        &sAfter,
+				FetchOwner:        &fetchOwner,
 			})
 		return SendXMLResponse(ctx, res, err,
 			&MetaOpts{

tests/integration/group-tests.go

@@ -218,6 +218,7 @@ func TestListObjects(s *S3Conf) {
 	ListObjects_marker_not_from_obj_list(s)
 	ListObjects_list_all_objs(s)
 	ListObjects_nested_dir_file_objs(s)
+	ListObjects_check_owner(s)
 	//TODO: remove the condition after implementing checksums in azure
 	if !s.azureTests {
 		ListObjects_with_checksum(s)
@@ -235,6 +236,7 @@ func TestListObjectsV2(s *S3Conf) {
 	ListObjectsV2_all_objs_max_keys(s)
 	ListObjectsV2_exceeding_max_keys(s)
 	ListObjectsV2_list_all_objs(s)
+	ListObjectsV2_with_owner(s)
 	//TODO: remove the condition after implementing checksums in azure
 	if !s.azureTests {
 		ListObjectsV2_with_checksum(s)
@@ -897,6 +899,7 @@ func GetIntTests() IntTests {
 		"ListObjects_marker_not_from_obj_list":                                    ListObjects_marker_not_from_obj_list,
 		"ListObjects_list_all_objs":                                               ListObjects_list_all_objs,
 		"ListObjects_nested_dir_file_objs":                                        ListObjects_nested_dir_file_objs,
+		"ListObjects_check_owner":                                                 ListObjects_check_owner,
 		"ListObjects_with_checksum":                                               ListObjects_with_checksum,
 		"ListObjectsV2_start_after":                                               ListObjectsV2_start_after,
 		"ListObjectsV2_both_start_after_and_continuation_token":                   ListObjectsV2_both_start_after_and_continuation_token,
@@ -907,6 +910,7 @@ func GetIntTests() IntTests {
 		"ListObjectsV2_truncated_common_prefixes":                                 ListObjectsV2_truncated_common_prefixes,
 		"ListObjectsV2_all_objs_max_keys":                                         ListObjectsV2_all_objs_max_keys,
 		"ListObjectsV2_list_all_objs":                                             ListObjectsV2_list_all_objs,
+		"ListObjectsV2_with_owner":                                                ListObjectsV2_with_owner,
 		"ListObjectsV2_with_checksum":                                             ListObjectsV2_with_checksum,
 		"ListObjectVersions_VD_success":                                           ListObjectVersions_VD_success,
 		"DeleteObject_non_existing_object":                                        DeleteObject_non_existing_object,

tests/integration/tests.go

@@ -4815,7 +4815,7 @@ func ListObjects_with_checksum(s *S3Conf) error {
 			return err
 		}
 
-		if !compareObjects(res.Contents, contents) {
+		if !compareObjects(contents, res.Contents) {
 			return fmt.Errorf("expected the objects list to be %v, instead got %v", contents, res.Contents)
 		}
 
@@ -4899,6 +4899,38 @@ func ListObjects_nested_dir_file_objs(s *S3Conf) error {
 	})
 }
 
+func ListObjects_check_owner(s *S3Conf) error {
+	testName := "ListObjects_check_owner"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		objs, err := putObjects(s3client, []string{"foo", "bar/baz", "quxx/xyz/eee", "abc/", "bcc"}, bucket)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.ListObjects(ctx, &s3.ListObjectsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		for i := range res.Contents {
+			res.Contents[i].Owner = &types.Owner{
+				ID: &s.awsID,
+			}
+		}
+
+		if !compareObjects(objs, res.Contents) {
+			return fmt.Errorf("expected the contents to be %v, instead got %v", objs, res.Contents)
+		}
+
+		return nil
+
+	})
+}
+
 func ListObjectsV2_start_after(s *S3Conf) error {
 	testName := "ListObjectsV2_start_after"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -5290,6 +5322,38 @@ func ListObjectsV2_list_all_objs(s *S3Conf) error {
 	})
 }
 
+func ListObjectsV2_with_owner(s *S3Conf) error {
+	testName := "ListObjectsV2_with_owner"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		objs, err := putObjects(s3client, []string{"foo", "bar/baz", "quxx/xyz/eee", "abc/", "bcc"}, bucket)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+			Bucket:     &bucket,
+			FetchOwner: getBoolPtr(true),
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		for i := range res.Contents {
+			res.Contents[i].Owner = &types.Owner{
+				ID: &s.awsID,
+			}
+		}
+
+		if !compareObjects(objs, res.Contents) {
+			return fmt.Errorf("expected the contents to be %v, instead got %v", objs, res.Contents)
+		}
+
+		return nil
+	})
+}
+
 func ListObjectsV2_with_checksum(s *S3Conf) error {
 	testName := "ListObjectsV2_with_checksum"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {

tests/integration/utils.go

@@ -725,10 +725,16 @@ func compareObjects(list1, list2 []types.Object) bool {
 		if obj.ChecksumType != "" {
 			if obj.ChecksumType[0] != list2[i].ChecksumType[0] {
 				fmt.Printf("checksum types are not equal: (%q %q) %v != %v\n",
-					*obj.Key, *list2[i].Key, obj.ChecksumType[0], list2[i].ChecksumType[0])
+					*obj.Key, *list2[i].Key, obj.ChecksumType, list2[i].ChecksumType)
 				return false
 			}
 		}
+		if obj.Owner != nil {
+			if *obj.Owner.ID != *list2[i].Owner.ID {
+				fmt.Printf("object owner IDs not equal: (%q %q) %v != %v\n",
+					*obj.Key, *list2[i].Key, *obj.Owner.ID, *list2[i].Owner.ID)
+			}
+		}
 	}
 
 	return true