fix: isolate Snowball replication trust per entry

Evaluate PutObject and ReplicateObject permissions with immutable per-entry request snapshots during concurrent Snowball extraction. Preserve the first API error without sharing mutable handler state, and cover prefix-scoped trust under the race detector.

Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/auth-handler.go

@@ -786,10 +786,20 @@ func isPutActionAllowedWithRequestTags(ctx context.Context, atype authType, buck
 		return s3Err
 	}
 
-	logger.GetReqInfo(ctx).Cred = cred
-	logger.GetReqInfo(ctx).Owner = owner
-	logger.GetReqInfo(ctx).Region = region
+	reqInfo := logger.GetReqInfo(ctx)
+	if reqInfo == nil {
+		return ErrAccessDenied
+	}
+	reqInfo.Lock()
+	reqInfo.Cred = cred
+	reqInfo.Owner = owner
+	reqInfo.Region = region
+	reqInfo.Unlock()
 
+	return isPutActionAllowedWithCred(bucketName, objectName, r, action, requestTags, cred, owner)
+}
+
+func isPutActionAllowedWithCred(bucketName, objectName string, r *http.Request, action policy.Action, requestTags *string, cred auth.Credentials, owner bool) APIErrorCode {
 	// Do not check for PutObjectRetentionAction permission,
 	// if mode and retain until date are not set.
 	// Can happen when bucket has default lock config set

cmd/object-handlers.go

@@ -2491,6 +2491,11 @@ func (api objectAPIHandlers) PutObjectExtractHandler(w http.ResponseWriter, r *h
 			sha256hex = getContentSha256Cksum(r, serviceS3)
 		}
 	}
+	entryRequestBase := r.Clone(ctx)
+	// The streaming reader fills r.Trailer while untar writes small entries in
+	// parallel. Entry authorization never consumes trailers, so keep them out
+	// of the immutable request template cloned by those goroutines.
+	entryRequestBase.Trailer = nil
 
 	hreader, err := hash.NewReader(ctx, reader, size, md5hex, sha256hex, size)
 	if err != nil {
@@ -2515,9 +2520,9 @@ func (api objectAPIHandlers) PutObjectExtractHandler(w http.ResponseWriter, r *h
 	rawReplica := hasReplicaStatus(r.Header)
 	markerExact := hasReplicationMarker(r.Header)
 	trustedRequestCtx := withReplicationTrust(ctx, true, rawReplica)
-	trustedRequest := r.WithContext(trustedRequestCtx)
+	trustedRequest := entryRequestBase.WithContext(trustedRequestCtx)
 	cleanRequestCtx := withReplicationTrust(ctx, false, false)
-	cleanRequest := cloneRequestWithoutReplicationHeaders(r, cleanRequestCtx)
+	cleanRequest := cloneRequestWithoutReplicationHeaders(entryRequestBase, cleanRequestCtx)
 	trustedReqParams := extractReqParams(trustedRequest)
 	cleanReqParams := extractReqParams(cleanRequest)
 
@@ -2533,29 +2538,44 @@ func (api objectAPIHandlers) PutObjectExtractHandler(w http.ResponseWriter, r *h
 	if sc == "" {
 		sc = storageclass.STANDARD
 	}
+	reqInfo := logger.GetReqInfo(ctx)
+	if reqInfo == nil {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
+		return
+	}
+	reqInfo.RLock()
+	tarCred := reqInfo.Cred
+	tarOwner := reqInfo.Owner
+	reqInfo.RUnlock()
+	var tarS3Err atomic.Int32
+	setTarS3Err := func(code APIErrorCode) {
+		tarS3Err.CompareAndSwap(int32(ErrNone), int32(code))
+	}
 
 	putObjectTar := func(reader io.Reader, info os.FileInfo, object string) error {
 		size := info.Size()
-		if s3Err = isPutActionAllowed(ctx, getRequestAuthType(r), bucket, object, r, policy.PutObjectAction); s3Err != ErrNone {
-			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
-			return errors.New(errorCodes.ToAPIErr(s3Err).Code)
+		entryAuthReq := entryRequestBase.Clone(ctx)
+		entryS3Err := isPutActionAllowedWithCred(bucket, object, entryAuthReq, policy.PutObjectAction, nil, tarCred, tarOwner)
+		if entryS3Err != ErrNone {
+			setTarS3Err(entryS3Err)
+			return errors.New(errorCodes.ToAPIErr(entryS3Err).Code)
 		}
 		replicationPermitted := false
-		if rawReplica || markerExact {
-			replicationPermitted = replicationPermissionAllowed(ctx, r, bucket, object, policy.ReplicateObjectAction)
+		if tarCred.AccessKey != "" && (rawReplica || markerExact) {
+			replicationPermitted = isPutActionAllowedWithCred(bucket, object, entryAuthReq, policy.ReplicateObjectAction, nil, tarCred, tarOwner) == ErrNone
 		}
 		if rawReplica && !replicationPermitted {
-			s3Err = ErrAccessDenied
-			return errors.New(errorCodes.ToAPIErr(s3Err).Code)
+			setTarS3Err(ErrAccessDenied)
+			return errors.New(errorCodes.ToAPIErr(ErrAccessDenied).Code)
 		}
 		entryTrusted := markerExact && replicationPermitted
 		replicaTrusted := entryTrusted && rawReplica
 		entryCtx := cleanRequestCtx
-		entryReq := cleanRequest
+		entryReq := cloneRequestWithoutReplicationHeaders(entryAuthReq, cleanRequestCtx)
 		reqParams := cleanReqParams
 		if entryTrusted {
 			entryCtx = trustedRequestCtx
-			entryReq = trustedRequest
+			entryReq = entryAuthReq.WithContext(trustedRequestCtx)
 			reqParams = trustedReqParams
 		}
 		metadata := map[string]string{
@@ -2592,7 +2612,7 @@ func (api objectAPIHandlers) PutObjectExtractHandler(w http.ResponseWriter, r *h
 		pReader := NewPutObjReader(rawReader)
 
 		if replicaTrusted {
-			if err = extractReplicationMetadataFromMime(entryCtx, textproto.MIMEHeader(entryReq.Header), metadata); err != nil {
+			if err := extractReplicationMetadataFromMime(entryCtx, textproto.MIMEHeader(entryReq.Header), metadata); err != nil {
 				return err
 			}
 			metadata[xhttp.AmzBucketReplicationStatus] = replication.Replica.String()
@@ -2657,7 +2677,7 @@ func (api objectAPIHandlers) PutObjectExtractHandler(w http.ResponseWriter, r *h
 		}
 
 		if s3err != ErrNone {
-			s3Err = s3err
+			setTarS3Err(s3err)
 			return ObjectLocked{}
 		}
 
@@ -2746,7 +2766,14 @@ func (api objectAPIHandlers) PutObjectExtractHandler(w http.ResponseWriter, r *h
 		return nil
 	}
 
-	if err = untar(ctx, hreader, putObjectTar, opts); err != nil {
+	err = untar(ctx, hreader, putObjectTar, opts)
+	if code := APIErrorCode(tarS3Err.Load()); code != ErrNone {
+		s3Err = code
+		if err == nil {
+			err = errors.New(errorCodes.ToAPIErr(code).Code)
+		}
+	}
+	if err != nil {
 		apiErr := errorCodes.ToAPIErr(s3Err)
 		// If not set, convert or use BadRequest
 		if s3Err == ErrNone {

cmd/replication-trust.go

@@ -106,11 +106,9 @@ func hasReplicationRequestHeaders(h http.Header) bool {
 }
 
 func cloneRequestWithoutReplicationHeaders(r *http.Request, ctx context.Context) *http.Request {
-	clone := new(http.Request)
-	*clone = *r
-	clone.Header = r.Header.Clone()
+	clone := r.Clone(ctx)
 	stripReplicationRequestHeaders(clone.Header)
-	return clone.WithContext(ctx)
+	return clone
 }
 
 // applyReplicationTrust binds the handler context to the effective request.

cmd/replication-trust_test.go

@@ -11,6 +11,7 @@
 package cmd
 
 import (
+	"archive/tar"
 	"bytes"
 	"context"
 	"crypto/md5"
@@ -21,11 +22,14 @@ import (
 	"net/http/httptest"
 	"net/url"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/minio/madmin-go/v3"
 	"github.com/minio/minio/internal/auth"
 	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/pkg/v3/policy"
 )
 
 func TestAPIReplicationTrustProtectsSSECReads(t *testing.T) {
@@ -272,6 +276,153 @@ func testAPIPutObjectReplicationTrust(obj ObjectLayer, instanceType, bucketName
 	}
 }
 
+func TestAPISnowballReplicationTrustIsPerEntry(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:          t,
+		objAPITest: testAPISnowballReplicationTrustIsPerEntry,
+	})
+}
+
+func testAPISnowballReplicationTrustIsPerEntry(obj ObjectLayer, instanceType, bucketName string,
+	apiRouter http.Handler, _ auth.Credentials, t *testing.T,
+) {
+	const (
+		allowedPrefix = "snowball/allowed/"
+		deniedPrefix  = "snowball/denied/"
+		sourceETag    = "0123456789abcdef0123456789abcdef"
+	)
+	creds := newSnowballReplicationTrustUser(t, instanceType, bucketName, allowedPrefix)
+
+	var body bytes.Buffer
+	tw := tar.NewWriter(&body)
+	objects := make([]struct {
+		name    string
+		trusted bool
+	}, 0, 32)
+	for i := 0; i < 16; i++ {
+		for _, entry := range []struct {
+			prefix  string
+			trusted bool
+		}{
+			{prefix: allowedPrefix, trusted: true},
+			{prefix: deniedPrefix},
+		} {
+			name := entry.prefix + strconv.Itoa(i)
+			data := []byte("snowball replication trust " + name)
+			if err := tw.WriteHeader(&tar.Header{Name: name, Mode: 0o600, Size: int64(len(data))}); err != nil {
+				t.Fatal(err)
+			}
+			if _, err := tw.Write(data); err != nil {
+				t.Fatal(err)
+			}
+			objects = append(objects, struct {
+				name    string
+				trusted bool
+			}{name: name, trusted: entry.trusted})
+		}
+	}
+	if err := tw.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	headers := map[string]string{
+		xhttp.AmzSnowballExtract:            "true",
+		xhttp.MinIOSourceReplicationRequest: "true",
+		xhttp.MinIOSourceETag:               sourceETag,
+	}
+	for _, test := range []struct {
+		name    string
+		trailer bool
+	}{
+		{name: "signed-v4"},
+		{name: "streaming-unsigned-trailer", trailer: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			var req *http.Request
+			var err error
+			if test.trailer {
+				req, err = newStreamingUnsignedTrailerRequest(http.MethodPut,
+					getPutObjectURL("", bucketName, "snowball.tar"), body.Bytes(), UTCNow())
+				if err == nil {
+					for name, value := range headers {
+						req.Header.Set(name, value)
+					}
+					err = signRequestV4(req, creds.AccessKey, creds.SecretKey)
+				}
+			} else {
+				req, err = newTestSignedRequestV4(http.MethodPut, getPutObjectURL("", bucketName, "snowball.tar"),
+					int64(body.Len()), bytes.NewReader(body.Bytes()), creds.AccessKey, creds.SecretKey, headers)
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			rec := httptest.NewRecorder()
+			apiRouter.ServeHTTP(rec, req)
+			if rec.Code != http.StatusOK {
+				t.Fatalf("%s: Snowball PUT status %d: %s", instanceType, rec.Code, rec.Body.String())
+			}
+
+			for _, object := range objects {
+				info, err := obj.GetObjectInfo(t.Context(), bucketName, object.name, ObjectOptions{})
+				if err != nil {
+					t.Fatalf("%s: get %s: %v", instanceType, object.name, err)
+				}
+				if object.trusted && info.ETag != sourceETag {
+					t.Errorf("%s: trusted entry %s ETag = %q, want source ETag", instanceType, object.name, info.ETag)
+				}
+				if !object.trusted && info.ETag == sourceETag {
+					t.Errorf("%s: untrusted entry %s preserved source ETag", instanceType, object.name)
+				}
+			}
+		})
+	}
+}
+
+func newSnowballReplicationTrustUser(t *testing.T, instanceType, bucketName, allowedPrefix string) auth.Credentials {
+	t.Helper()
+	ctx := t.Context()
+	accessKey, secretKey, err := auth.GenerateCredentials()
+	if err != nil {
+		t.Fatalf("%s: generate credentials: %v", instanceType, err)
+	}
+	creds := auth.Credentials{AccessKey: accessKey, SecretKey: secretKey}
+	if _, err = globalIAMSys.CreateUser(ctx, creds.AccessKey, madmin.AddOrUpdateUserReq{
+		SecretKey: creds.SecretKey,
+		Status:    madmin.AccountEnabled,
+	}); err != nil {
+		t.Fatalf("%s: create Snowball user: %v", instanceType, err)
+	}
+
+	policyJSON := `{
+ "Version": "2012-10-17",
+ "Statement": [
+  {
+   "Effect": "Allow",
+   "Action": ["s3:PutObject"],
+   "Resource": ["arn:aws:s3:::` + bucketName + `/*"]
+  },
+  {
+   "Effect": "Allow",
+   "Action": ["s3:ReplicateObject"],
+   "Resource": ["arn:aws:s3:::` + bucketName + `/` + allowedPrefix + `*"]
+  }
+ ]
+}`
+	parsed, err := policy.ParseConfig(strings.NewReader(policyJSON))
+	if err != nil {
+		t.Fatalf("%s: parse Snowball policy: %v", instanceType, err)
+	}
+	policyName := "snowball-replication-trust-" + mustGetUUID()
+	if _, err = globalIAMSys.SetPolicy(ctx, policyName, *parsed); err != nil {
+		t.Fatalf("%s: install Snowball policy: %v", instanceType, err)
+	}
+	if _, err = globalIAMSys.PolicyDBSet(ctx, creds.AccessKey, policyName, regUser, false); err != nil {
+		t.Fatalf("%s: attach Snowball policy: %v", instanceType, err)
+	}
+	return creds
+}
+
 func TestAPICopyObjectMarkerOnlyDoesNotCopyCiphertext(t *testing.T) {
 	defer DetectTestLeak(t)()
 	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{