astelm/minio
1
0
Fork 0
mirror of https://github.com/pgsty/minio.git synced 2026-09-05 18:16:16 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix: keep streaming trailers visible after replication headers are stripped

Browse Source 
A request that does not earn replication trust continues with a clone whose
internal replication headers are removed. r.Clone copies the Trailer map, but
the streaming body reader created from the original request fills the
original map, so a trailing checksum was never seen by the hash reader and
PutObject and UploadPart with STREAMING-UNSIGNED-PAYLOAD-TRAILER failed with
XAmzContentChecksumMismatch whenever an untrusted X-Minio-Source-* header was
present. Share the trailer map with the clone, as the Snowball path already
does for its per-entry requests, and cover both handlers with a test.

The marker evaluation that was copied into six handlers now lives in
evaluateReplicationTrust so the rule (a declared replica without the
replication permission is rejected; trust needs the exact marker plus the
permission) is defined once.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PvgysXDmhPBBimCReYtA8q
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
2026-09-02 14:06:56 +08:00
parent 6586fbfd0d
commit 76195f1c68
4 changed files with 95 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/object-handlers.go
+9 -30
View File
@@ -1308,18 +1308,11 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidStorageClass), r.URL)
return
}
rawReplica := hasReplicaStatus(r.Header)
markerExact := hasReplicationMarker(r.Header)
replicationPermitted := false
if rawReplica || markerExact {
replicationPermitted = replicationPermissionAllowed(ctx, r, dstBucket, dstObject, policy.ReplicateObjectAction)
}
if rawReplica && !replicationPermitted {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
trustedReplication, replicaTrusted, trustErr := evaluateReplicationTrust(ctx, r, dstBucket, dstObject, policy.ReplicateObjectAction)
if trustErr != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(trustErr), r.URL)
return
}
trustedReplication := markerExact && replicationPermitted
replicaTrusted := trustedReplication && rawReplica
if hasReplicationRequestHeaders(r.Header) {
ctx, r = applyReplicationTrust(ctx, r, trustedReplication, replicaTrusted)
}
@@ -2073,14 +2066,9 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
}
}
rawReplica := hasReplicaStatus(r.Header)
markerExact := hasReplicationMarker(r.Header)
replicationPermitted := false
if rawReplica || markerExact {
replicationPermitted = replicationPermissionAllowed(ctx, r, bucket, object, policy.ReplicateObjectAction)
}
if rawReplica && !replicationPermitted {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
trustedReplication, replicaTrusted, trustErr := evaluateReplicationTrust(ctx, r, bucket, object, policy.ReplicateObjectAction)
if trustErr != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(trustErr), r.URL)
return
}
if _, ok := r.Header[xhttp.MinIOSourceReplicationCheck]; ok {
@@ -2093,8 +2081,6 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
trustedReplication := markerExact && replicationPermitted
replicaTrusted := trustedReplication && rawReplica
if hasReplicationRequestHeaders(r.Header) {
ctx, r = applyReplicationTrust(ctx, r, trustedReplication, replicaTrusted)
}
@@ -2832,17 +2818,11 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
rawReplica := hasReplicaStatus(r.Header)
markerExact := hasReplicationMarker(r.Header)
replicationPermitted := false
if rawReplica || markerExact {
replicationPermitted = replicationPermissionAllowed(ctx, r, bucket, object, policy.ReplicateDeleteAction)
}
if rawReplica && !replicationPermitted {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
trustedReplication, replica, trustErr := evaluateReplicationTrust(ctx, r, bucket, object, policy.ReplicateDeleteAction)
if trustErr != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(trustErr), r.URL)
return
}
trustedReplication := markerExact && replicationPermitted
var s3Error APIErrorCode
if trustedReplication {
s3Error = authorizeReplicationDelete(ctx, r)
@@ -2858,7 +2838,6 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrReplicationPermissionCheckError), r.URL)
return
}
replica := trustedReplication && rawReplica
if hasReplicationRequestHeaders(r.Header) {
ctx, r = applyReplicationTrust(ctx, r, trustedReplication, replica)
}
cmd/object-multipart-handlers.go
+10 -29
View File
@@ -171,18 +171,11 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
rawReplica := hasReplicaStatus(r.Header)
markerExact := hasReplicationMarker(r.Header)
replicationPermitted := false
if rawReplica || markerExact {
replicationPermitted = replicationPermissionAllowed(ctx, r, bucket, object, policy.ReplicateObjectAction)
}
if rawReplica && !replicationPermitted {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
trustedReplication, replicaTrusted, trustErr := evaluateReplicationTrust(ctx, r, bucket, object, policy.ReplicateObjectAction)
if trustErr != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(trustErr), r.URL)
return
}
trustedReplication := markerExact && replicationPermitted
replicaTrusted := trustedReplication && rawReplica
if hasReplicationRequestHeaders(r.Header) {
ctx, r = applyReplicationTrust(ctx, r, trustedReplication, replicaTrusted)
}
@@ -870,17 +863,11 @@ func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
rawReplica := hasReplicaStatus(r.Header)
markerExact := hasReplicationMarker(r.Header)
replicationPermitted := false
if rawReplica || markerExact {
replicationPermitted = replicationPermissionAllowed(ctx, r, bucket, object, policy.ReplicateObjectAction)
}
if rawReplica && !replicationPermitted {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
trustedReplication, _, trustErr := evaluateReplicationTrust(ctx, r, bucket, object, policy.ReplicateObjectAction)
if trustErr != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(trustErr), r.URL)
return
}
trustedReplication := markerExact && replicationPermitted
storedReplica := mi.UserDefined[xhttp.AmzBucketReplicationStatus] == replication.Replica.String()
replicaTrusted := trustedReplication && storedReplica
if hasReplicationRequestHeaders(r.Header) {
@@ -1110,19 +1097,13 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
rawReplica := hasReplicaStatus(r.Header)
markerExact := hasReplicationMarker(r.Header)
replicationPermitted := false
if rawReplica || markerExact {
replicationPermitted = replicationPermissionAllowed(ctx, r, bucket, object, policy.ReplicateObjectAction)
}
if rawReplica && !replicationPermitted {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
trustedReplication, replicaTrusted, trustErr := evaluateReplicationTrust(ctx, r, bucket, object, policy.ReplicateObjectAction)
if trustErr != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(trustErr), r.URL)
return
}
trustedReplication := markerExact && replicationPermitted
if hasReplicationRequestHeaders(r.Header) {
ctx, r = applyReplicationTrust(ctx, r, trustedReplication, trustedReplication && rawReplica)
ctx, r = applyReplicationTrust(ctx, r, trustedReplication, replicaTrusted)
}
// Get upload id.
cmd/replication-trust.go
+22
View File
@@ -70,6 +70,25 @@ func replicationPermissionAllowed(ctx context.Context, r *http.Request, bucket,
return authorizeRequest(ctx, r, action) == ErrNone
}
// evaluateReplicationTrust decides whether a request may carry replication
// semantics for the given action. A request that declares itself a replica
// without holding the replication permission is rejected. trusted reports that
// the exact marker came from a permitted principal; replica additionally
// requires the request to declare REPLICA status.
func evaluateReplicationTrust(ctx context.Context, r *http.Request, bucket, object string, action policy.Action) (trusted, replica bool, s3Err APIErrorCode) {
rawReplica := hasReplicaStatus(r.Header)
markerExact := hasReplicationMarker(r.Header)
permitted := false
if rawReplica || markerExact {
permitted = replicationPermissionAllowed(ctx, r, bucket, object, action)
}
if rawReplica && !permitted {
return false, false, ErrAccessDenied
}
trusted = markerExact && permitted
return trusted, trusted && rawReplica, ErrNone
}
// replicationRequestHeaders are internal request controls. They are removed
// only after signature verification when a request has not earned replication
// trust. Public S3/SSE/checksum headers, proxy loop guards, and replication
@@ -110,6 +129,9 @@ func hasReplicationRequestHeaders(h http.Header) bool {
func cloneRequestWithoutReplicationHeaders(ctx context.Context, r *http.Request) *http.Request {
clone := r.Clone(ctx)
stripReplicationRequestHeaders(clone.Header)
// A streaming body reader built from the original request fills r.Trailer
// as the body is consumed; the checksum reader must observe that same map.
clone.Trailer = r.Trailer
return clone
}
cmd/replication-trust_test.go
+54
View File
@@ -830,3 +830,57 @@ func testAPISSECMultipartReplicationTrust(obj ObjectLayer, instanceType, bucketN
t.Fatal("replicated SSE-C multipart object did not decrypt to source plaintext")
}
}
// TestAPIStreamingTrailerWithUntrustedReplicationHeaders verifies that a
// request which does not earn replication trust is still processed as an
// ordinary upload. The streaming body reader fills the original request's
// trailer while the handler continues with a header-stripped clone, so the
// trailing checksum must remain visible through that clone.
func TestAPIStreamingTrailerWithUntrustedReplicationHeaders(t *testing.T) {
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: testAPIStreamingTrailerWithUntrustedReplicationHeaders})
}
func testAPIStreamingTrailerWithUntrustedReplicationHeaders(obj ObjectLayer, instanceType, bucketName string, apiRouter http.Handler, _ auth.Credentials, t *testing.T) {
putOnly := newObjectAttributesAuthzUser(t, instanceType, bucketName, `"s3:PutObject"`)
payload := bytes.Repeat([]byte("trailer probe "), 4096)
send := func(targetURL string) *httptest.ResponseRecorder {
req, err := newStreamingUnsignedTrailerRequest(http.MethodPut, targetURL, payload, UTCNow())
if err != nil {
t.Fatal(err)
}
req.Header.Set(xhttp.MinIOSourceReplicationRequest, "true")
req.Header.Set(xhttp.MinIOSourceETag, "forged-etag")
if err := signRequestV4(req, putOnly.AccessKey, putOnly.SecretKey); err != nil {
t.Fatal(err)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
return rec
}
if rec := send(getPutObjectURL("", bucketName, "trailer-object")); rec.Code != http.StatusOK {
t.Fatalf("%s: PutObject status %d: %s", instanceType, rec.Code, rec.Body.String())
}
info, err := obj.GetObjectInfo(t.Context(), bucketName, "trailer-object", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
if info.Size != int64(len(payload)) || info.ETag == "forged-etag" {
t.Fatalf("%s: stored size %d etag %q, want %d bytes with a computed etag", instanceType, info.Size, info.ETag, len(payload))
}
upload, err := obj.NewMultipartUpload(t.Context(), bucketName, "trailer-multipart", ObjectOptions{})
if err != nil {
t.Fatal(err)
}
if rec := send(getPutObjectPartURL("", bucketName, "trailer-multipart", upload.UploadID, "1")); rec.Code != http.StatusOK {
t.Fatalf("%s: PutObjectPart status %d: %s", instanceType, rec.Code, rec.Body.String())
}
parts, err := obj.ListObjectParts(t.Context(), bucketName, "trailer-multipart", upload.UploadID, 0, 10, ObjectOptions{})
if err != nil {
t.Fatal(err)
}
if len(parts.Parts) != 1 || parts.Parts[0].Size != int64(len(payload)) {
t.Fatalf("%s: uploaded parts %+v, want one part of %d bytes", instanceType, parts.Parts, len(payload))
}
}