s3: apply destination retention on copies instead of inheriting it (#11773)

* s3: apply destination retention on copies instead of inheriting it

CopyObject carried the source object's retention mode, retain-until date
and legal hold into the destination entry, so a copy onto a bucket with
default COMPLIANCE retention produced an unlocked object that could be
deleted early. Copied objects now take their retention from the copy
request's lock headers or the destination bucket's default, matching
PUT semantics; a versioned self-copy gets the same treatment while an
unversioned metadata-only self-copy keeps the object's existing lock.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: validate lock headers on metadata-only self-copies

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/s3api/s3api_object_handlers_copy.go

@@ -352,6 +352,12 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 				updatedEntry = cloneProtoEntry(rawEntry)
 			}
 			updatedEntry.Extended = mergeCopyMetadata(updatedEntry.Extended, updatedMetadata)
+			// The update keeps the object's existing lock state, but request
+			// lock headers still need validating so a disallowed combination
+			// is rejected rather than silently ignored.
+			if errCode := s3a.applyCopyObjectLockHeaders(r, dstBucket, dstObject, updatedEntry); errCode != s3err.ErrNone {
+				return errCode
+			}
 			if updatedEntry.Attributes == nil {
 				updatedEntry.Attributes = &filer_pb.FuseAttributes{}
 			}
@@ -455,6 +461,12 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 		if k == s3_constants.ExtAmzOwnerKey || k == s3_constants.ExtAmzAclKey {
 			continue
 		}
+		// Retention and legal hold are destination settings, not object
+		// metadata: the copy takes its own lock headers or the destination
+		// bucket's default instead.
+		if isCopyLockKey(k) {
+			continue
+		}
 		// Skip encryption-specific headers that might conflict with destination encryption type
 		skipHeader := false
 
@@ -491,6 +503,11 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 	// so REPLACE doesn't leak source values through the merge, as on self-copies.
 	dstEntry.Extended = mergeCopyMetadata(dstEntry.Extended, processedMetadata)
 
+	if errCode := s3a.applyCopyObjectLockHeaders(r, dstBucket, dstObject, dstEntry); errCode != s3err.ErrNone {
+		s3err.WriteErrorResponse(w, r, errCode)
+		return
+	}
+
 	// For zero-size files or files without chunks, handle inline content
 	// This includes encrypted inline files that need decryption/re-encryption
 	if entry.Attributes.FileSize == 0 || len(entry.GetChunks()) == 0 {
@@ -599,6 +616,39 @@ func copyReplacesSourceEntry(sameDestination bool, dstVersioningState, srcVersio
 	return sameDestination && dstVersioningState == "" && srcVersionId == ""
 }
 
+// isCopyLockKey marks the extended attributes that carry object lock state; a
+// copy never inherits them from the source.
+func isCopyLockKey(key string) bool {
+	switch key {
+	case s3_constants.ExtObjectLockModeKey, s3_constants.ExtRetentionUntilDateKey, s3_constants.ExtLegalHoldKey:
+		return true
+	}
+	return false
+}
+
+// applyCopyObjectLockHeaders resolves the copied object's retention: explicit
+// copy-request lock headers win, then the destination bucket's default, and
+// the source's own retention never carries over.
+func (s3a *S3ApiServer) applyCopyObjectLockHeaders(r *http.Request, dstBucket, dstObject string, dstEntry *filer_pb.Entry) s3err.ErrorCode {
+	objectLockEnabled, lockErr := s3a.isObjectLockEnabled(dstBucket)
+	if lockErr != nil && !errors.Is(lockErr, filer_pb.ErrNotFound) {
+		glog.Errorf("CopyObjectHandler: failed to check object lock for bucket %s: %v", dstBucket, lockErr)
+		return s3err.ErrInternalError
+	}
+	if validationErr := s3a.validateObjectLockHeaders(r, dstBucket, dstObject, objectLockEnabled); validationErr != nil {
+		glog.V(2).Infof("CopyObjectHandler: object lock header validation failed for %s/%s: %v", dstBucket, dstObject, validationErr)
+		return mapValidationErrorToS3Error(validationErr)
+	}
+	if !objectLockEnabled {
+		return s3err.ErrNone
+	}
+	if err := s3a.extractObjectLockMetadataFromRequest(r, dstEntry); err != nil {
+		glog.Errorf("CopyObjectHandler: failed to extract object lock metadata for %s/%s: %v", dstBucket, dstObject, err)
+		return mapValidationErrorToS3Error(err)
+	}
+	return s3err.ErrNone
+}
+
 func cloneProtoEntry(entry *filer_pb.Entry) *filer_pb.Entry {
 	if entry == nil {
 		return nil

weed/s3api/s3api_object_lock_fix_test.go

@@ -79,3 +79,20 @@ func TestVeeamObjectLockBugFix(t *testing.T) {
 		assert.True(t, enabled, "Object Lock should be enabled when constant is used")
 	})
 }
+
+// Copied objects get their own retention from the request or the destination
+// bucket's default; the source's lock keys never carry into the destination
+// entry.
+func TestCopyDoesNotInheritLockKeys(t *testing.T) {
+	// The destination-Extended copy loop filters on isCopyLockKey, so
+	// retention and legal hold never carry over from the source entry.
+	for _, key := range []string{
+		s3_constants.ExtObjectLockModeKey,
+		s3_constants.ExtRetentionUntilDateKey,
+		s3_constants.ExtLegalHoldKey,
+	} {
+		assert.True(t, isCopyLockKey(key), "%s must be filtered from a copy destination", key)
+	}
+	assert.False(t, isCopyLockKey(s3_constants.ExtAmzOwnerKey))
+	assert.False(t, isCopyLockKey("custom-key"))
+}