From 0f5c6a8ae0b1866758ce956b784fbb44a1d38250 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 27 Mar 2026 12:42:57 -0700 Subject: [PATCH] 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 | 18 +++++++++ weed/s3api/s3api_object_handlers_put.go | 52 +++++++++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index ef9b58733..de5b924b0 100644 --- a/weed/s3api/filer_multipart.go +++ b/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) diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 2625df8be..dfbc9bdea 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/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 {