diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index c7b2400f5..e8d3a9083 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -51,6 +51,13 @@ func (s3a *S3ApiServer) createMultipartUpload(r *http.Request, input *s3.CreateM entry.Extended = make(map[string][]byte) } entry.Extended["key"] = []byte(*input.Key) + + // Set object owner for multipart upload + amzAccountId := r.Header.Get(s3_constants.AmzAccountId) + if amzAccountId != "" { + entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId) + } + for k, v := range input.Metadata { entry.Extended[k] = []byte(*v) } @@ -92,7 +99,7 @@ type CompleteMultipartUploadResult struct { VersionId *string `xml:"-"` } -func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploadInput, parts *CompleteMultipartUpload) (output *CompleteMultipartUploadResult, code s3err.ErrorCode) { +func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.CompleteMultipartUploadInput, parts *CompleteMultipartUpload) (output *CompleteMultipartUploadResult, code s3err.ErrorCode) { glog.V(2).Infof("completeMultipartUpload input %v", input) if len(parts.Parts) == 0 { @@ -254,6 +261,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa } versionEntry.Extended[s3_constants.ExtVersionIdKey] = []byte(versionId) versionEntry.Extended[s3_constants.SeaweedFSUploadId] = []byte(*input.UploadId) + + // Set object owner for versioned multipart objects + amzAccountId := r.Header.Get(s3_constants.AmzAccountId) + if amzAccountId != "" { + versionEntry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId) + } + for k, v := range pentry.Extended { if k != "key" { versionEntry.Extended[k] = v @@ -296,6 +310,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa entry.Extended = make(map[string][]byte) } entry.Extended[s3_constants.ExtVersionIdKey] = []byte("null") + + // Set object owner for suspended versioning multipart objects + amzAccountId := r.Header.Get(s3_constants.AmzAccountId) + if amzAccountId != "" { + entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId) + } + for k, v := range pentry.Extended { if k != "key" { entry.Extended[k] = v @@ -329,6 +350,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa entry.Extended = make(map[string][]byte) } entry.Extended[s3_constants.SeaweedFSUploadId] = []byte(*input.UploadId) + + // Set object owner for non-versioned multipart objects + amzAccountId := r.Header.Get(s3_constants.AmzAccountId) + if amzAccountId != "" { + entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId) + } + for k, v := range pentry.Extended { if k != "key" { entry.Extended[k] = v diff --git a/weed/s3api/s3api_bucket_config.go b/weed/s3api/s3api_bucket_config.go index 5987a9de6..463587255 100644 --- a/weed/s3api/s3api_bucket_config.go +++ b/weed/s3api/s3api_bucket_config.go @@ -98,10 +98,11 @@ func (s3a *S3ApiServer) getBucketConfig(bucket string) (*BucketConfig, s3err.Err return config, s3err.ErrNone } - // Load from filer - bucketEntry, err := s3a.getEntry(s3a.option.BucketsPath, bucket) + // Try to get from filer + entry, err := s3a.getEntry(s3a.option.BucketsPath, bucket) if err != nil { - if err == filer_pb.ErrNotFound { + if errors.Is(err, filer_pb.ErrNotFound) { + // Bucket doesn't exist return nil, s3err.ErrNoSuchBucket } glog.Errorf("getBucketConfig: failed to get bucket entry for %s: %v", bucket, err) @@ -110,25 +111,25 @@ func (s3a *S3ApiServer) getBucketConfig(bucket string) (*BucketConfig, s3err.Err config := &BucketConfig{ Name: bucket, - Entry: bucketEntry, + Entry: entry, } // Extract configuration from extended attributes - if bucketEntry.Extended != nil { - if versioning, exists := bucketEntry.Extended[s3_constants.ExtVersioningKey]; exists { + if entry.Extended != nil { + if versioning, exists := entry.Extended[s3_constants.ExtVersioningKey]; exists { config.Versioning = string(versioning) } - if ownership, exists := bucketEntry.Extended[s3_constants.ExtOwnershipKey]; exists { + if ownership, exists := entry.Extended[s3_constants.ExtOwnershipKey]; exists { config.Ownership = string(ownership) } - if acl, exists := bucketEntry.Extended[s3_constants.ExtAmzAclKey]; exists { + if acl, exists := entry.Extended[s3_constants.ExtAmzAclKey]; exists { config.ACL = acl } - if owner, exists := bucketEntry.Extended[s3_constants.ExtAmzOwnerKey]; exists { + if owner, exists := entry.Extended[s3_constants.ExtAmzOwnerKey]; exists { config.Owner = string(owner) } // Parse Object Lock configuration if present - if objectLockConfig, found := LoadObjectLockConfigurationFromExtended(bucketEntry); found { + if objectLockConfig, found := LoadObjectLockConfigurationFromExtended(entry); found { config.ObjectLockConfig = objectLockConfig glog.V(2).Infof("getBucketConfig: cached Object Lock configuration for bucket %s", bucket) } diff --git a/weed/s3api/s3api_bucket_handlers.go b/weed/s3api/s3api_bucket_handlers.go index bc8ef574b..410e8aa3d 100644 --- a/weed/s3api/s3api_bucket_handlers.go +++ b/weed/s3api/s3api_bucket_handlers.go @@ -230,7 +230,7 @@ func (s3a *S3ApiServer) HeadBucketHandler(w http.ResponseWriter, r *http.Request bucket, _ := s3_constants.GetBucketAndObject(r) glog.V(3).Infof("HeadBucketHandler %s", bucket) - if entry, err := s3a.getEntry(s3a.option.BucketsPath, bucket); entry == nil || err == filer_pb.ErrNotFound { + if entry, err := s3a.getEntry(s3a.option.BucketsPath, bucket); entry == nil || errors.Is(err, filer_pb.ErrNotFound) { s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket) return } @@ -240,7 +240,7 @@ func (s3a *S3ApiServer) HeadBucketHandler(w http.ResponseWriter, r *http.Request func (s3a *S3ApiServer) checkBucket(r *http.Request, bucket string) s3err.ErrorCode { entry, err := s3a.getEntry(s3a.option.BucketsPath, bucket) - if entry == nil || err == filer_pb.ErrNotFound { + if entry == nil || errors.Is(err, filer_pb.ErrNotFound) { return s3err.ErrNoSuchBucket } @@ -669,7 +669,7 @@ func (s3a *S3ApiServer) DeleteBucketOwnershipControls(w http.ResponseWriter, r * bucketEntry, err := s3a.getEntry(s3a.option.BucketsPath, bucket) if err != nil { - if err == filer_pb.ErrNotFound { + if errors.Is(err, filer_pb.ErrNotFound) { s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket) return } diff --git a/weed/s3api/s3api_object_handlers_acl.go b/weed/s3api/s3api_object_handlers_acl.go new file mode 100644 index 000000000..7185f9896 --- /dev/null +++ b/weed/s3api/s3api_object_handlers_acl.go @@ -0,0 +1,236 @@ +package s3api + +import ( + "context" + "errors" + "fmt" + "net/http" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" +) + +// GetObjectAclHandler Get object ACL +// https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObjectAcl.html +func (s3a *S3ApiServer) GetObjectAclHandler(w http.ResponseWriter, r *http.Request) { + // collect parameters + bucket, object := s3_constants.GetBucketAndObject(r) + glog.V(3).Infof("GetObjectAclHandler %s %s", bucket, object) + + if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, err) + return + } + + // Check if object exists and get its metadata + bucketDir := s3a.option.BucketsPath + "/" + bucket + entry, err := s3a.getEntry(bucketDir, object) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey) + return + } + glog.Errorf("GetObjectAclHandler: error checking object %s/%s: %v", bucket, object, err) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return + } + if entry == nil { + s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey) + return + } + + // Get object owner from metadata, fallback to request account + var objectOwner string + var objectOwnerDisplayName string + amzAccountId := r.Header.Get(s3_constants.AmzAccountId) + + if entry.Extended != nil { + if ownerBytes, exists := entry.Extended[s3_constants.ExtAmzOwnerKey]; exists { + objectOwner = string(ownerBytes) + } + } + + // Fallback to current account if no owner stored + if objectOwner == "" { + objectOwner = amzAccountId + } + + objectOwnerDisplayName = s3a.iam.GetAccountNameById(objectOwner) + + // Build ACL response + response := AccessControlPolicy{ + Owner: CanonicalUser{ + ID: objectOwner, + DisplayName: objectOwnerDisplayName, + }, + } + + // Get grants from stored ACL metadata + grants := GetAcpGrants(entry.Extended) + if len(grants) > 0 { + // Convert AWS SDK grants to local Grant format + for _, grant := range grants { + localGrant := Grant{ + Permission: Permission(*grant.Permission), + } + + if grant.Grantee != nil { + localGrant.Grantee = Grantee{ + Type: *grant.Grantee.Type, + XMLXSI: "CanonicalUser", + XMLNS: "http://www.w3.org/2001/XMLSchema-instance", + } + + if grant.Grantee.ID != nil { + localGrant.Grantee.ID = *grant.Grantee.ID + localGrant.Grantee.DisplayName = s3a.iam.GetAccountNameById(*grant.Grantee.ID) + } + + if grant.Grantee.URI != nil { + localGrant.Grantee.URI = *grant.Grantee.URI + } + } + + response.AccessControlList.Grant = append(response.AccessControlList.Grant, localGrant) + } + } else { + // Fallback to default full control for object owner + response.AccessControlList.Grant = append(response.AccessControlList.Grant, Grant{ + Grantee: Grantee{ + ID: objectOwner, + DisplayName: objectOwnerDisplayName, + Type: "CanonicalUser", + XMLXSI: "CanonicalUser", + XMLNS: "http://www.w3.org/2001/XMLSchema-instance"}, + Permission: Permission(s3_constants.PermissionFullControl), + }) + } + + writeSuccessResponseXML(w, r, response) +} + +// PutObjectAclHandler Put object ACL +// https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectAcl.html +func (s3a *S3ApiServer) PutObjectAclHandler(w http.ResponseWriter, r *http.Request) { + // collect parameters + bucket, object := s3_constants.GetBucketAndObject(r) + glog.V(3).Infof("PutObjectAclHandler %s %s", bucket, object) + + if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, err) + return + } + + // Check if object exists and get its metadata + bucketDir := s3a.option.BucketsPath + "/" + bucket + entry, err := s3a.getEntry(bucketDir, object) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey) + return + } + glog.Errorf("PutObjectAclHandler: error checking object %s/%s: %v", bucket, object, err) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return + } + if entry == nil { + s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey) + return + } + + // Get current object owner from metadata + var objectOwner string + amzAccountId := r.Header.Get(s3_constants.AmzAccountId) + + if entry.Extended != nil { + if ownerBytes, exists := entry.Extended[s3_constants.ExtAmzOwnerKey]; exists { + objectOwner = string(ownerBytes) + } + } + + // Fallback to current account if no owner stored + if objectOwner == "" { + objectOwner = amzAccountId + } + + // **PERMISSION CHECKS** + + // 1. Check if user is admin (admins can modify any ACL) + if !s3a.isUserAdmin(r) { + // 2. Check object ownership - only object owner can modify ACL (unless admin) + if objectOwner != amzAccountId { + glog.V(3).Infof("PutObjectAclHandler: Access denied - user %s is not owner of object %s/%s (owner: %s)", + amzAccountId, bucket, object, objectOwner) + s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) + return + } + + // 3. Check object-level WRITE_ACP permission + // Create the specific action for this object + writeAcpAction := Action(fmt.Sprintf("WriteAcp:%s/%s", bucket, object)) + identity, errCode := s3a.iam.authRequest(r, writeAcpAction) + if errCode != s3err.ErrNone { + glog.V(3).Infof("PutObjectAclHandler: Auth failed for WriteAcp action on %s/%s: %v", bucket, object, errCode) + s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) + return + } + + // 4. Verify the authenticated identity can perform WriteAcp on this specific object + if identity == nil || !identity.canDo(writeAcpAction, bucket, object) { + glog.V(3).Infof("PutObjectAclHandler: Identity %v cannot perform WriteAcp on %s/%s", identity, bucket, object) + s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) + return + } + } else { + glog.V(3).Infof("PutObjectAclHandler: Admin user %s granted ACL modification permission for %s/%s", amzAccountId, bucket, object) + } + + // Get bucket config for ownership settings + bucketConfig, errCode := s3a.getBucketConfig(bucket) + if errCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, errCode) + return + } + + bucketOwnership := bucketConfig.Ownership + bucketOwnerId := bucketConfig.Owner + + // Extract ACL from request (either canned ACL or XML body) + // This function also validates that the owner in the request matches the object owner + grants, errCode := ExtractAcl(r, s3a.iam, bucketOwnership, bucketOwnerId, objectOwner, amzAccountId) + if errCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, errCode) + return + } + + // Store ACL in object metadata + if errCode := AssembleEntryWithAcp(entry, objectOwner, grants); errCode != s3err.ErrNone { + glog.Errorf("PutObjectAclHandler: failed to assemble entry with ACP: %v", errCode) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return + } + + // Update the object with new ACL metadata + err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + request := &filer_pb.UpdateEntryRequest{ + Directory: bucketDir, + Entry: entry, + } + + if _, err := client.UpdateEntry(context.Background(), request); err != nil { + return err + } + return nil + }) + + if err != nil { + glog.Errorf("PutObjectAclHandler: failed to update entry: %v", err) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return + } + + glog.V(3).Infof("PutObjectAclHandler: Successfully updated ACL for %s/%s by user %s", bucket, object, amzAccountId) + writeSuccessResponseEmpty(w, r) +} diff --git a/weed/s3api/s3api_object_handlers_multipart.go b/weed/s3api/s3api_object_handlers_multipart.go index a6be3b4a1..9d8411883 100644 --- a/weed/s3api/s3api_object_handlers_multipart.go +++ b/weed/s3api/s3api_object_handlers_multipart.go @@ -3,6 +3,7 @@ package s3api import ( "crypto/sha1" "encoding/xml" + "errors" "fmt" "io" "net/http" @@ -41,7 +42,7 @@ func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http // Check if versioning is enabled for the bucket (needed for object lock) versioningEnabled, err := s3a.isVersioningEnabled(bucket) if err != nil { - if err == filer_pb.ErrNotFound { + if errors.Is(err, filer_pb.ErrNotFound) { s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket) return } @@ -111,7 +112,7 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r return } - response, errCode := s3a.completeMultipartUpload(&s3.CompleteMultipartUploadInput{ + response, errCode := s3a.completeMultipartUpload(r, &s3.CompleteMultipartUploadInput{ Bucket: aws.String(bucket), Key: objectKey(aws.String(object)), UploadId: aws.String(uploadID), diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 4114fe8f2..23069ef32 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -98,7 +98,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) // Get detailed versioning state for the bucket versioningState, err := s3a.getVersioningState(bucket) if err != nil { - if err == filer_pb.ErrNotFound { + if errors.Is(err, filer_pb.ErrNotFound) { s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket) return } @@ -213,6 +213,14 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, uploadUrl string, dataReader proxyReq.Header.Add(header, value) } } + + // Set object owner header for filer to extract + amzAccountId := r.Header.Get(s3_constants.AmzAccountId) + if amzAccountId != "" { + proxyReq.Header.Set(s3_constants.ExtAmzOwnerKey, amzAccountId) + glog.V(2).Infof("putToFiler: setting owner header %s for object %s", amzAccountId, uploadUrl) + } + // ensure that the Authorization header is overriding any previous // Authorization header which might be already present in proxyReq s3a.maybeAddFilerJwtAuthorization(proxyReq, true) @@ -244,8 +252,8 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, uploadUrl string, dataReader glog.Errorf("upload to filer error: %v", ret.Error) return "", filerErrorToS3Error(ret.Error) } + stats_collect.RecordBucketActiveTime(bucket) - stats_collect.S3BucketTrafficReceivedBytesCounter.WithLabelValues(bucket).Add(float64(ret.Size)) return etag, s3err.ErrNone } @@ -290,6 +298,18 @@ func (s3a *S3ApiServer) maybeGetFilerJwtAuthorizationToken(isWrite bool) string return string(encodedJwt) } +// setObjectOwnerFromRequest sets the object owner metadata based on the authenticated user +func (s3a *S3ApiServer) setObjectOwnerFromRequest(r *http.Request, entry *filer_pb.Entry) { + amzAccountId := r.Header.Get(s3_constants.AmzAccountId) + if amzAccountId != "" { + if entry.Extended == nil { + entry.Extended = make(map[string][]byte) + } + entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId) + glog.V(2).Infof("setObjectOwnerFromRequest: set object owner to %s", amzAccountId) + } +} + // putVersionedObject handles PUT operations for versioned buckets using the new layout // where all versions (including latest) are stored in the .versions directory func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, object string, dataReader io.Reader, objectContentType string) (etag string, errCode s3err.ErrorCode) { @@ -321,6 +341,9 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob } entry.Extended[s3_constants.ExtVersionIdKey] = []byte("null") + // Set object owner for suspended versioning objects + s3a.setObjectOwnerFromRequest(r, entry) + // Extract and store object lock metadata from request headers (if any) if err := s3a.extractObjectLockMetadataFromRequest(r, entry); err != nil { glog.Errorf("putSuspendedVersioningObject: failed to extract object lock metadata: %v", err) @@ -466,6 +489,9 @@ func (s3a *S3ApiServer) putVersionedObject(r *http.Request, bucket, object strin } versionEntry.Extended[s3_constants.ExtETagKey] = []byte(etag) + // Set object owner for versioned objects + s3a.setObjectOwnerFromRequest(r, versionEntry) + // Extract and store object lock metadata from request headers if err := s3a.extractObjectLockMetadataFromRequest(r, versionEntry); err != nil { glog.Errorf("putVersionedObject: failed to extract object lock metadata: %v", err) diff --git a/weed/s3api/s3api_object_handlers_skip.go b/weed/s3api/s3api_object_handlers_skip.go deleted file mode 100644 index 0b74a0ec7..000000000 --- a/weed/s3api/s3api_object_handlers_skip.go +++ /dev/null @@ -1,21 +0,0 @@ -package s3api - -import ( - "net/http" -) - -// GetObjectAclHandler Get object ACL -// https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObjectAcl.html -func (s3a *S3ApiServer) GetObjectAclHandler(w http.ResponseWriter, r *http.Request) { - - w.WriteHeader(http.StatusNoContent) - -} - -// PutObjectAclHandler Put object ACL -// https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectAcl.html -func (s3a *S3ApiServer) PutObjectAclHandler(w http.ResponseWriter, r *http.Request) { - - w.WriteHeader(http.StatusNoContent) - -} diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index a0d02d566..ff82969b5 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/weed/s3api/s3api_object_versioning.go @@ -278,7 +278,7 @@ func (s3a *S3ApiServer) findVersionsRecursively(currentPath, relativePath string VersionId: version.VersionId, IsLatest: version.IsLatest, LastModified: version.LastModified, - Owner: CanonicalUser{ID: "unknown", DisplayName: "unknown"}, + Owner: s3a.getObjectOwnerFromVersion(version, bucket, objectKey), } *allVersions = append(*allVersions, deleteMarker) } else { @@ -289,7 +289,7 @@ func (s3a *S3ApiServer) findVersionsRecursively(currentPath, relativePath string LastModified: version.LastModified, ETag: version.ETag, Size: version.Size, - Owner: CanonicalUser{ID: "unknown", DisplayName: "unknown"}, + Owner: s3a.getObjectOwnerFromVersion(version, bucket, objectKey), StorageClass: "STANDARD", } *allVersions = append(*allVersions, versionEntry) @@ -339,7 +339,7 @@ func (s3a *S3ApiServer) findVersionsRecursively(currentPath, relativePath string LastModified: time.Unix(entry.Attributes.Mtime, 0), ETag: etag, Size: int64(entry.Attributes.FileSize), - Owner: CanonicalUser{ID: "unknown", DisplayName: "unknown"}, + Owner: s3a.getObjectOwnerFromEntry(entry), StorageClass: "STANDARD", } *allVersions = append(*allVersions, versionEntry) @@ -761,3 +761,55 @@ func (s3a *S3ApiServer) getLatestObjectVersion(bucket, object string) (*filer_pb return latestVersionEntry, nil } + +// getObjectOwnerFromVersion extracts object owner information from version entry metadata +func (s3a *S3ApiServer) getObjectOwnerFromVersion(version *ObjectVersion, bucket, objectKey string) CanonicalUser { + // First try to get owner from the version entry itself + if version.Entry != nil && version.Entry.Extended != nil { + if ownerBytes, exists := version.Entry.Extended[s3_constants.ExtAmzOwnerKey]; exists { + ownerId := string(ownerBytes) + ownerDisplayName := s3a.iam.GetAccountNameById(ownerId) + return CanonicalUser{ID: ownerId, DisplayName: ownerDisplayName} + } + } + + // Fallback: try to get owner from the current version of the object + // This handles cases where older versions might not have owner metadata + if version.VersionId == "null" { + // For null version, check the regular object file + bucketDir := s3a.option.BucketsPath + "/" + bucket + if entry, err := s3a.getEntry(bucketDir, objectKey); err == nil && entry.Extended != nil { + if ownerBytes, exists := entry.Extended[s3_constants.ExtAmzOwnerKey]; exists { + ownerId := string(ownerBytes) + ownerDisplayName := s3a.iam.GetAccountNameById(ownerId) + return CanonicalUser{ID: ownerId, DisplayName: ownerDisplayName} + } + } + } else { + // For versioned objects, try to get from latest version metadata + if latestVersion, err := s3a.getLatestObjectVersion(bucket, objectKey); err == nil && latestVersion.Extended != nil { + if ownerBytes, exists := latestVersion.Extended[s3_constants.ExtAmzOwnerKey]; exists { + ownerId := string(ownerBytes) + ownerDisplayName := s3a.iam.GetAccountNameById(ownerId) + return CanonicalUser{ID: ownerId, DisplayName: ownerDisplayName} + } + } + } + + // Ultimate fallback: return anonymous if no owner found + return CanonicalUser{ID: s3_constants.AccountAnonymousId, DisplayName: "anonymous"} +} + +// getObjectOwnerFromEntry extracts object owner information from a file entry +func (s3a *S3ApiServer) getObjectOwnerFromEntry(entry *filer_pb.Entry) CanonicalUser { + if entry != nil && entry.Extended != nil { + if ownerBytes, exists := entry.Extended[s3_constants.ExtAmzOwnerKey]; exists { + ownerId := string(ownerBytes) + ownerDisplayName := s3a.iam.GetAccountNameById(ownerId) + return CanonicalUser{ID: ownerId, DisplayName: ownerDisplayName} + } + } + + // Fallback: return anonymous if no owner found + return CanonicalUser{ID: s3_constants.AccountAnonymousId, DisplayName: "anonymous"} +}