fix S3 conditional header TOCTOU race during concurrent writes (#8800)

The early check in checkConditionalHeaders ran before data upload,
allowing concurrent requests to all pass before any write completed.

Use the existing distributed lock manager to atomically re-check
conditional headers and create the filer entry. The lock is acquired
after data upload (so it's held only for fast metadata operations)
and covers all conditional headers: If-Match, If-None-Match,
If-Modified-Since, and If-Unmodified-Since.

Requests without conditional headers have zero overhead (no lock).

weed/s3api/filer_multipart.go

@@ -457,6 +457,15 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 		}
 	} else if vErr == nil && versioningState == s3_constants.VersioningSuspended {
 		// For suspended versioning, add "null" version ID metadata and return "null" version ID
+		// If conditional headers are present, acquire a distributed lock to atomically
+		// re-check conditions and create the entry, preventing TOCTOU races.
+		condLock, condErr := s3a.lockAndRecheckConditionalHeaders(r, *input.Bucket, *input.Key)
+		if condErr != s3err.ErrNone {
+			return nil, condErr
+		}
+		if condLock != nil {
+			defer condLock.StopShortLivedLock()
+		}
 		err = s3a.mkFile(dirName, entryName, finalParts, func(entry *filer_pb.Entry) {
 			if entry.Extended == nil {
 				entry.Extended = make(map[string][]byte)
@@ -512,6 +521,15 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 		}
 	} else {
 		// For non-versioned buckets, create main object file
+		// If conditional headers are present, acquire a distributed lock to atomically
+		// re-check conditions and create the entry, preventing TOCTOU races.
+		condLock, condErr := s3a.lockAndRecheckConditionalHeaders(r, *input.Bucket, *input.Key)
+		if condErr != s3err.ErrNone {
+			return nil, condErr
+		}
+		if condLock != nil {
+			defer condLock.StopShortLivedLock()
+		}
 		err = s3a.mkFile(dirName, entryName, finalParts, func(entry *filer_pb.Entry) {
 			if entry.Extended == nil {
 				entry.Extended = make(map[string][]byte)

weed/s3api/s3api_object_handlers_put.go

@@ -17,6 +17,7 @@ import (
 	"time"
 
 	"github.com/pquerna/cachecontrol/cacheobject"
+	"github.com/seaweedfs/seaweedfs/weed/cluster"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/operation"
@@ -697,6 +698,19 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 	// Step 4: Save metadata to filer via gRPC
 	// Use context.Background() to ensure metadata save completes even if HTTP request is cancelled
 	// This matches the chunk upload behavior and prevents orphaned chunks
+	//
+	// If conditional headers are present, acquire a distributed lock to atomically
+	// re-check conditions and create the entry. This prevents TOCTOU races where
+	// concurrent requests all pass the early check before any write completes.
+	condBucket, condObject := s3_constants.GetBucketAndObject(r)
+	condLock, condErr := s3a.lockAndRecheckConditionalHeaders(r, condBucket, condObject)
+	if condErr != s3err.ErrNone {
+		s3a.deleteOrphanedChunks(chunkResult.FileChunks)
+		return "", condErr, SSEResponseMetadata{}
+	}
+	if condLock != nil {
+		defer condLock.StopShortLivedLock()
+	}
 	glog.V(3).Infof("putToFiler: About to create entry - dir=%s, name=%s, chunks=%d, extended keys=%d",
 		path.Dir(filePath), path.Base(filePath), len(entry.Chunks), len(entry.Extended))
 	createErr := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
@@ -1822,6 +1836,44 @@ func (s3a *S3ApiServer) checkConditionalHeaders(r *http.Request, bucket, object
 	return s3a.validateConditionalHeaders(r, headers, entry, bucket, object)
 }
 
+// lockAndRecheckConditionalHeaders acquires a distributed lock and re-checks conditional
+// headers atomically before a write operation. This prevents TOCTOU races where concurrent
+// requests could all pass the early (unlocked) check before any write completes.
+//
+// Returns (nil, ErrNone) if no conditional headers are present (no lock needed).
+// Returns (lock, ErrNone) if the re-check passes — the caller MUST defer lock.StopShortLivedLock()
+// to hold the lock across the subsequent write and release it after.
+// Returns (nil, errCode) if the re-check fails — no lock to release.
+func (s3a *S3ApiServer) lockAndRecheckConditionalHeaders(r *http.Request, bucket, object string) (*cluster.LiveLock, s3err.ErrorCode) {
+	headers, errCode := parseConditionalHeaders(r)
+	if errCode != s3err.ErrNone {
+		return nil, errCode
+	}
+	if !headers.isSet {
+		return nil, s3err.ErrNone
+	}
+
+	lockKey := s3a.toFilerPath(bucket, object)
+	lockClient := cluster.NewLockClient(s3a.option.GrpcDialOption, s3a.option.Filers[0])
+	lock := lockClient.NewShortLivedLock(lockKey, fmt.Sprintf("s3-cond-%d", s3a.randomClientId))
+
+	entry, err := s3a.resolveObjectEntry(bucket, object)
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			entry = nil
+		} else {
+			lock.StopShortLivedLock()
+			glog.Errorf("lockAndRecheckConditionalHeaders: error resolving object entry for %s/%s: %v", bucket, object, err)
+			return nil, s3err.ErrInternalError
+		}
+	}
+	if errCode = s3a.validateConditionalHeaders(r, headers, entry, bucket, object); errCode != s3err.ErrNone {
+		lock.StopShortLivedLock()
+		return nil, errCode
+	}
+	return lock, s3err.ErrNone
+}
+
 // validateConditionalHeadersForReads checks conditional headers for read operations against the provided entry
 func (s3a *S3ApiServer) validateConditionalHeadersForReads(r *http.Request, headers conditionalHeaders, entry *filer_pb.Entry, bucket, object string) ConditionalHeaderResult {
 	if !headers.isSet {