Object locking need to persist the tags and set the headers (#6994)

* fix object locking read and write

No logic to include object lock metadata in HEAD/GET response headers
No logic to extract object lock metadata from PUT request headers

* add tests for object locking

* Update weed/s3api/s3api_object_handlers_put.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update weed/s3api/s3api_object_handlers.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* refactor

* add unit tests

* sync versions

* Update s3_worm_integration_test.go

* fix legal hold values

* lint

* fix tests

* racing condition when enable versioning

* fix tests

* validate put object lock header

* allow check lock permissions for PUT

* default to OFF legal hold

* only set object lock headers for objects that are actually from object lock-enabled buckets

fix     --- FAIL: TestAddObjectLockHeadersToResponse/Handle_entry_with_no_object_lock_metadata (0.00s)

* address comments

* fix tests

* purge

* fix

* refactoring

* address comment

* address comment

* Update weed/s3api/s3api_object_handlers_put.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update weed/s3api/s3api_object_handlers_put.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update weed/s3api/s3api_object_handlers.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* avoid nil

* ensure locked objects cannot be overwritten

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

1 files changed, 160 insertions, 4 deletions

diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go
index 29c31b6bd..ebdfc8567 100644
--- a/weed/s3api/s3api_object_handlers_put.go
+++ b/weed/s3api/s3api_object_handlers_put.go
@@ -3,9 +3,11 @@ package s3api
 import (
 	"crypto/md5"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
@@ -20,6 +22,18 @@ import (
 	stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
 )
 
+// Object lock validation errors
+var (
+	ErrObjectLockVersioningRequired       = errors.New("object lock headers can only be used on versioned buckets")
+	ErrInvalidObjectLockMode              = errors.New("invalid object lock mode")
+	ErrInvalidLegalHoldStatus             = errors.New("invalid legal hold status")
+	ErrInvalidRetentionDateFormat         = errors.New("invalid retention until date format")
+	ErrRetentionDateMustBeFuture          = errors.New("retention until date must be in the future")
+	ErrObjectLockModeRequiresDate         = errors.New("object lock mode requires retention until date")
+	ErrRetentionDateRequiresMode          = errors.New("retention until date requires object lock mode")
+	ErrGovernanceBypassVersioningRequired = errors.New("governance bypass header can only be used on versioned buckets")
+)
+
 func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
 
 	// http://docs.aws.amazon.com/AmazonS3/latest/dev/UploadingObjects.html
@@ -85,13 +99,24 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 
 		glog.V(1).Infof("PutObjectHandler: bucket %s, object %s, versioningEnabled=%v", bucket, object, versioningEnabled)
 
-		// Check object lock permissions before PUT operation (only for versioned buckets)
-		bypassGovernance := r.Header.Get("x-amz-bypass-governance-retention") == "true"
-		if err := s3a.checkObjectLockPermissionsForPut(r, bucket, object, bypassGovernance, versioningEnabled); err != nil {
-			s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
+		// Validate object lock headers before processing
+		if err := s3a.validateObjectLockHeaders(r, versioningEnabled); err != nil {
+			glog.V(2).Infof("PutObjectHandler: object lock header validation failed for bucket %s, object %s: %v", bucket, object, err)
+			s3err.WriteErrorResponse(w, r, mapValidationErrorToS3Error(err))
 			return
 		}
 
+		// For non-versioned buckets, check if existing object has object lock protections
+		// that would prevent overwrite (PUT operations overwrite existing objects in non-versioned buckets)
+		if !versioningEnabled {
+			bypassGovernance := r.Header.Get("x-amz-bypass-governance-retention") == "true"
+			if err := s3a.checkObjectLockPermissions(r, bucket, object, "", bypassGovernance); err != nil {
+				glog.V(2).Infof("PutObjectHandler: object lock permissions check failed for %s/%s: %v", bucket, object, err)
+				s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
+				return
+			}
+		}
+
 		if versioningEnabled {
 			// Handle versioned PUT
 			glog.V(1).Infof("PutObjectHandler: using versioned PUT for %s/%s", bucket, object)
@@ -287,6 +312,12 @@ func (s3a *S3ApiServer) putVersionedObject(r *http.Request, bucket, object strin
 	}
 	versionEntry.Extended[s3_constants.ExtETagKey] = []byte(etag)
 
+	// 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)
+		return "", "", s3err.ErrInvalidRequest
+	}
+
 	// Update the version entry with metadata
 	err = s3a.mkFile(bucketDir, versionObjectPath, versionEntry.Chunks, func(updatedEntry *filer_pb.Entry) {
 		updatedEntry.Extended = versionEntry.Extended
@@ -341,3 +372,128 @@ func (s3a *S3ApiServer) updateLatestVersionInDirectory(bucket, object, versionId
 
 	return nil
 }
+
+// extractObjectLockMetadataFromRequest extracts object lock headers from PUT requests
+// and stores them in the entry's Extended attributes
+func (s3a *S3ApiServer) extractObjectLockMetadataFromRequest(r *http.Request, entry *filer_pb.Entry) error {
+	if entry.Extended == nil {
+		entry.Extended = make(map[string][]byte)
+	}
+
+	// Extract object lock mode (GOVERNANCE or COMPLIANCE)
+	if mode := r.Header.Get(s3_constants.AmzObjectLockMode); mode != "" {
+		entry.Extended[s3_constants.ExtObjectLockModeKey] = []byte(mode)
+		glog.V(2).Infof("extractObjectLockMetadataFromRequest: storing object lock mode: %s", mode)
+	}
+
+	// Extract retention until date
+	if retainUntilDate := r.Header.Get(s3_constants.AmzObjectLockRetainUntilDate); retainUntilDate != "" {
+		// Parse the ISO8601 date and convert to Unix timestamp for storage
+		parsedTime, err := time.Parse(time.RFC3339, retainUntilDate)
+		if err != nil {
+			glog.Errorf("extractObjectLockMetadataFromRequest: failed to parse retention until date, expected format: %s, error: %v", time.RFC3339, err)
+			return ErrInvalidRetentionDateFormat
+		}
+		entry.Extended[s3_constants.ExtRetentionUntilDateKey] = []byte(strconv.FormatInt(parsedTime.Unix(), 10))
+		glog.V(2).Infof("extractObjectLockMetadataFromRequest: storing retention until date (timestamp: %d)", parsedTime.Unix())
+	}
+
+	// Extract legal hold status
+	if legalHold := r.Header.Get(s3_constants.AmzObjectLockLegalHold); legalHold != "" {
+		// Store S3 standard "ON"/"OFF" values directly
+		if legalHold == s3_constants.LegalHoldOn || legalHold == s3_constants.LegalHoldOff {
+			entry.Extended[s3_constants.ExtLegalHoldKey] = []byte(legalHold)
+			glog.V(2).Infof("extractObjectLockMetadataFromRequest: storing legal hold: %s", legalHold)
+		} else {
+			glog.Errorf("extractObjectLockMetadataFromRequest: unexpected legal hold value provided, expected 'ON' or 'OFF'")
+			return ErrInvalidLegalHoldStatus
+		}
+	}
+
+	return nil
+}
+
+// validateObjectLockHeaders validates object lock headers in PUT requests
+func (s3a *S3ApiServer) validateObjectLockHeaders(r *http.Request, versioningEnabled bool) error {
+	// Extract object lock headers from request
+	mode := r.Header.Get(s3_constants.AmzObjectLockMode)
+	retainUntilDateStr := r.Header.Get(s3_constants.AmzObjectLockRetainUntilDate)
+	legalHold := r.Header.Get(s3_constants.AmzObjectLockLegalHold)
+
+	// Check if any object lock headers are present
+	hasObjectLockHeaders := mode != "" || retainUntilDateStr != "" || legalHold != ""
+
+	// Object lock headers can only be used on versioned buckets
+	if hasObjectLockHeaders && !versioningEnabled {
+		return ErrObjectLockVersioningRequired
+	}
+
+	// Validate object lock mode if present
+	if mode != "" {
+		if mode != s3_constants.RetentionModeGovernance && mode != s3_constants.RetentionModeCompliance {
+			return ErrInvalidObjectLockMode
+		}
+	}
+
+	// Validate retention date if present
+	if retainUntilDateStr != "" {
+		retainUntilDate, err := time.Parse(time.RFC3339, retainUntilDateStr)
+		if err != nil {
+			return ErrInvalidRetentionDateFormat
+		}
+
+		// Retention date must be in the future
+		if retainUntilDate.Before(time.Now()) {
+			return ErrRetentionDateMustBeFuture
+		}
+	}
+
+	// If mode is specified, retention date must also be specified
+	if mode != "" && retainUntilDateStr == "" {
+		return ErrObjectLockModeRequiresDate
+	}
+
+	// If retention date is specified, mode must also be specified
+	if retainUntilDateStr != "" && mode == "" {
+		return ErrRetentionDateRequiresMode
+	}
+
+	// Validate legal hold if present
+	if legalHold != "" {
+		if legalHold != s3_constants.LegalHoldOn && legalHold != s3_constants.LegalHoldOff {
+			return ErrInvalidLegalHoldStatus
+		}
+	}
+
+	// Check for governance bypass header - only valid for versioned buckets
+	bypassGovernance := r.Header.Get("x-amz-bypass-governance-retention") == "true"
+
+	// Governance bypass headers are only valid for versioned buckets (like object lock headers)
+	if bypassGovernance && !versioningEnabled {
+		return ErrGovernanceBypassVersioningRequired
+	}
+
+	return nil
+}
+
+// mapValidationErrorToS3Error maps object lock validation errors to appropriate S3 error codes
+func mapValidationErrorToS3Error(err error) s3err.ErrorCode {
+	switch {
+	case errors.Is(err, ErrObjectLockVersioningRequired):
+		return s3err.ErrInvalidRequest
+	case errors.Is(err, ErrInvalidObjectLockMode):
+		return s3err.ErrInvalidRequest
+	case errors.Is(err, ErrInvalidLegalHoldStatus):
+		return s3err.ErrInvalidRequest
+	case errors.Is(err, ErrInvalidRetentionDateFormat):
+		return s3err.ErrMalformedDate
+	case errors.Is(err, ErrRetentionDateMustBeFuture),
+		errors.Is(err, ErrObjectLockModeRequiresDate),
+		errors.Is(err, ErrRetentionDateRequiresMode):
+		return s3err.ErrInvalidRequest
+	case errors.Is(err, ErrGovernanceBypassVersioningRequired):
+		return s3err.ErrInvalidRequest
+	default:
+		return s3err.ErrInvalidRequest
+	}
+}