diff --git a/buildscripts/rebrand-guard/compat-baseline.json b/buildscripts/rebrand-guard/compat-baseline.json index b8468abc1..3b280e3dc 100644 --- a/buildscripts/rebrand-guard/compat-baseline.json +++ b/buildscripts/rebrand-guard/compat-baseline.json @@ -627,6 +627,7 @@ "x-minio-internal-transitioned-object", "x-minio-internal-xyz", "x-minio-key", + "x-minio-last-modified", "x-minio-lifecycleconfig-updatedat", "x-minio-meta", "x-minio-meta-appid", @@ -1071,6 +1072,7 @@ "cmd/metrics.go=\"Version of current MinIO server instance\"", "cmd/metrics.go=\"minio\"", "cmd/object-api-utils.go=\".minio.sys\"", + "cmd/object-handlers-common.go=\"X-Minio-Last-Modified\"", "cmd/object-handlers.go=\"minio-federated\"", "cmd/object-handlers.go=\"minio.metadata.\"", "cmd/object-handlers.go=\"minio.versionId\"", diff --git a/cmd/object-copy-federation-result_test.go b/cmd/object-copy-federation-result_test.go index 20d70f10b..4ad122202 100644 --- a/cmd/object-copy-federation-result_test.go +++ b/cmd/object-copy-federation-result_test.go @@ -5,12 +5,14 @@ package cmd import ( "bytes" + "encoding/xml" "net/http" "net/url" "strings" "testing" "time" + "github.com/minio/minio/internal/amztime" "github.com/minio/minio/internal/auth" sse "github.com/minio/minio/internal/bucket/encryption" "github.com/minio/minio/internal/event" @@ -69,6 +71,13 @@ func TestAPIFederatedCopyObjectVersionAndEvent(t *testing.T) { if err != nil || info.VersionID != versionID { t.Fatalf("response does not name the written version: %v, %q", err, info.VersionID) } + var response CopyObjectResponse + if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if want := amztime.ISO8601Format(info.ModTime.UTC()); response.LastModified != want { + t.Errorf("copy time = %q, want written version time %q", response.LastModified, want) + } } select { case evt := <-events: diff --git a/cmd/object-copy-federation_test.go b/cmd/object-copy-federation_test.go index 5c11863a8..8ab966dba 100644 --- a/cmd/object-copy-federation_test.go +++ b/cmd/object-copy-federation_test.go @@ -36,6 +36,7 @@ import ( miniogo "github.com/minio/minio-go/v7" miniocredentials "github.com/minio/minio-go/v7/pkg/credentials" "github.com/minio/minio-go/v7/pkg/set" + "github.com/minio/minio/internal/amztime" "github.com/minio/minio/internal/auth" "github.com/minio/minio/internal/config/dns" "github.com/minio/minio/internal/crypto" @@ -164,7 +165,7 @@ func setupCopyObjectFederationRemote(t *testing.T, objectAPI ObjectLayer, apiRou core, err := miniogo.NewCore(host, &miniogo.Options{ Creds: miniocredentials.NewStaticV4(cred.AccessKey, cred.SecretKey, ""), Secure: true, - Transport: transport, + Transport: federatedWriteTransport{transport}, }) if err != nil { return nil, err @@ -209,6 +210,50 @@ func federatedCopyRequest(t *testing.T, apiRouter http.Handler, credentials auth return rec } +func TestAPIFederatedCopyObjectRejectsRawSSECReplica(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + endpoints: []string{"CopyObject", "PutObject"}, + objAPITest: func(obj ObjectLayer, instanceType, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) { + remoteBucket, capture, cleanup := setupCopyObjectFederationRemote(t, obj, router, instanceType, bucket, true) + defer cleanup() + for _, body := range []string{"", "raw SSE-C replica"} { + for _, withKey := range []bool{false, true} { + t.Run("size="+strconv.Itoa(len(body))+"/key="+strconv.FormatBool(withKey), func(t *testing.T) { + putCopyChecksumSource(t, router, cred, bucket, "source", []byte(body), federationSSEHeaders("c", 0x11, false)) + headers := map[string]string{ + xhttp.MinIOSourceReplicationRequest: "true", + xhttp.AmzBucketReplicationStatus: "REPLICA", + } + if withKey { + maps.Copy(headers, federationSSEHeaders("c", 0x11, true)) + } + destination := "destination-" + strconv.Itoa(len(body)) + "-" + strconv.FormatBool(withKey) + rec := federatedCopyRequest(t, router, cred, bucket, "source", remoteBucket, destination, headers) + var response APIErrorResponse + if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if rec.Code != http.StatusNotImplemented || response.Code != "NotImplemented" || + !strings.Contains(response.Message, "federated raw SSE-C replica CopyObject") { + t.Errorf("copy = %d %s, want explicit 501 rejection", rec.Code, rec.Body.String()) + } + if _, err := obj.GetObjectInfo(t.Context(), remoteBucket, destination, ObjectOptions{}); !isErrObjectNotFound(err) { + t.Errorf("rejected copy created a destination: %v", err) + } + }) + } + } + capture.mu.Lock() + defer capture.mu.Unlock() + if len(capture.headers) != 0 { + t.Errorf("rejected copies made %d remote requests", len(capture.headers)) + } + }, + }) +} + // TestAPIFederatedCopyObjectInlineSource drives the legacy etcd federation // branch of CopyObjectHandler end to end for a source object stored inline. // Such a source carries x-minio-internal-inline-data in its stored metadata, @@ -504,7 +549,7 @@ const federationTestKMSKeyID = "federation-test-key" // federationSSEHeaders returns the request headers that select one server-side // encryption kind: "plain", "s3", "kms", "kms-context" (SSE-KMS with an -// explicit encryption context) or "c". SSE-C keys derive from keyByte so a +// explicit encryption context), "kms-empty-context" or "c". SSE-C keys derive from keyByte so a // case can name two distinct customer keys. With copySource the SSE-C key is // returned in its x-amz-copy-source-* form, the only kind a copy has to name // for its source; the other kinds then return nothing. @@ -517,11 +562,14 @@ func federationSSEHeaders(kind string, keyByte byte, copySource bool) map[string case "plain": case "s3": h[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionAES - case "kms", "kms-context": + case "kms", "kms-context", "kms-empty-context": h[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionKMS h[xhttp.AmzServerSideEncryptionKmsID] = federationTestKMSKeyID - if kind == "kms-context" { + switch kind { + case "kms-context": h[xhttp.AmzServerSideEncryptionKmsContext] = base64.StdEncoding.EncodeToString([]byte(`{"tenant":"federation"}`)) + case "kms-empty-context": + h[xhttp.AmzServerSideEncryptionKmsContext] = base64.StdEncoding.EncodeToString([]byte(`{}`)) } case "c": key := bytes.Repeat([]byte{keyByte}, 32) @@ -569,6 +617,77 @@ func federationGetObject(t *testing.T, apiRouter http.Handler, credentials auth. return rec } +// assertFederationCopyReadback checks storage and the public GET/HEAD checksum +// surface with the destination key, including the committed write's time. +func assertFederationCopyReadback(t *testing.T, obj ObjectLayer, router http.Handler, cred auth.Credentials, + bucket, object string, rec *httptest.ResponseRecorder, typ hash.ChecksumType, data []byte, compressed bool, keyHeaders map[string]string, +) ObjectInfo { + t.Helper() + assertCopyChecksumResponse(t, rec, typ, data) + headers := make(http.Header) + for k, v := range keyHeaders { + headers.Set(k, v) + } + info := assertCopyChecksum(t, obj, bucket, object, typ, data, compressed, headers) + var response CopyObjectResponse + if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if want := amztime.ISO8601Format(info.ModTime.UTC()); info.ModTime.IsZero() || response.LastModified != want { + t.Errorf("copy time = %q, want stored time %q", response.LastModified, want) + } + readHeaders := map[string]string{xhttp.AmzChecksumMode: "ENABLED"} + maps.Copy(readHeaders, keyHeaders) + for _, method := range []string{http.MethodGet, http.MethodHead} { + req, err := newTestSignedRequestV4(method, getGetObjectURL("", bucket, object), 0, nil, cred.AccessKey, cred.SecretKey, readHeaders) + if err != nil { + t.Fatal(err) + } + got := httptest.NewRecorder() + router.ServeHTTP(got, req) + if got.Code != http.StatusOK { + t.Fatalf("destination %s = %d %s", method, got.Code, got.Body.String()) + } + if method == http.MethodGet && !bytes.Equal(got.Body.Bytes(), data) { + t.Fatalf("destination GET differs from the %d-byte plaintext", len(data)) + } + if got.Header().Get(xhttp.ContentLength) != strconv.Itoa(len(data)) || + got.Header().Get(typ.Key()) != mustChecksum(t, typ, data) || + got.Header().Get(xhttp.AmzChecksumType) != xhttp.AmzChecksumTypeFullObject { + t.Fatalf("destination %s length/checksum headers = %v", method, got.Header()) + } + } + return info +} + +func assertFederationKMSContext(t *testing.T, info ObjectInfo, requested string) { + t.Helper() + decode := func(encoded string) map[string]string { + t.Helper() + value, err := base64.StdEncoding.DecodeString(encoded) + if err != nil { + t.Fatal(err) + } + var context map[string]string + if err := json.Unmarshal(value, &context); err != nil { + t.Fatal(err) + } + return context + } + want := map[string]string{} + if requested != "" { + want = decode(requested) + } + stored, present := info.UserDefined[crypto.MetaContext] + if len(want) == 0 { + if present { + t.Errorf("absent/empty destination context stored client metadata %q", stored) + } + } else if !present || !maps.Equal(decode(stored), want) { + t.Errorf("stored KMS context = %q, want %v", stored, want) + } +} + // TestAPIFederatedCopyObjectSSE guards the legacy etcd federation branch of // CopyObjectHandler for encrypted sources and destinations (#158). The proxy // reads its source through getObjectNInfo, which yields the decrypted and @@ -584,9 +703,8 @@ func TestAPIFederatedCopyObjectSSE(t *testing.T) { ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ t: t, objAPITest: testAPIFederatedCopyObjectSSE, - // The test router registers routes in this order, and the plain - // PutObject route has no query matcher, so the multipart routes must - // precede it or they would never be reached. + // Register PutObjectPart and CopyObject before the query-less + // PutObject PUT route; otherwise PutObject shadows them. endpoints: []string{ "NewMultipart", "PutObjectPart", "CompleteMultipart", "CopyObject", "PutObject", "HeadObject", "GetObject", @@ -595,7 +713,8 @@ func TestAPIFederatedCopyObjectSSE(t *testing.T) { } // putFederationMultipartSource uploads parts as one multipart object through -// the API router; headers apply to the NewMultipartUpload request. +// the API router. Headers select the upload's encryption/checksum; SSE-C keys +// and checksums are also supplied on the parts and completion as required. func putFederationMultipartSource(t *testing.T, apiRouter http.Handler, credentials auth.Credentials, bucket, object string, parts [][]byte, headers map[string]string, ) { @@ -619,15 +738,27 @@ func putFederationMultipartSource(t *testing.T, apiRouter http.Handler, credenti t.Fatalf("failed to decode NewMultipartUpload response: %v", err) } completion := CompleteMultipartUpload{} + partHeaders := map[string]string{} + for _, key := range []string{xhttp.AmzServerSideEncryptionCustomerAlgorithm, xhttp.AmzServerSideEncryptionCustomerKey, xhttp.AmzServerSideEncryptionCustomerKeyMD5} { + if value := headers[key]; value != "" { + partHeaders[key] = value + } + } + checksumType := hash.NewChecksumType(headers[xhttp.AmzChecksumAlgo], headers[xhttp.AmzChecksumType]) for i, part := range parts { - rec := do(http.MethodPut, getPutObjectPartURL("", bucket, object, upload.UploadID, strconv.Itoa(i+1)), part, nil) - completion.Parts = append(completion.Parts, CompletePart{PartNumber: i + 1, ETag: canonicalizeETag(rec.Header()[xhttp.ETag][0])}) + if checksumType.IsSet() { + partHeaders[checksumType.Key()] = mustChecksum(t, checksumType, part) + } + rec := do(http.MethodPut, getPutObjectPartURL("", bucket, object, upload.UploadID, strconv.Itoa(i+1)), part, partHeaders) + completion.Parts = append(completion.Parts, completePartWithChecksum(checksumType, i+1, + canonicalizeETag(rec.Header()[xhttp.ETag][0]), partHeaders[checksumType.Key()])) } body, err := xml.Marshal(completion) if err != nil { t.Fatalf("failed to encode CompleteMultipartUpload: %v", err) } - do(http.MethodPost, getCompleteMultipartUploadURL("", bucket, object, upload.UploadID), body, nil) + delete(partHeaders, checksumType.Key()) + do(http.MethodPost, getCompleteMultipartUploadURL("", bucket, object, upload.UploadID), body, partHeaders) } func testAPIFederatedCopyObjectSSE(objectAPI ObjectLayer, instanceType, bucketName string, @@ -652,9 +783,13 @@ func testAPIFederatedCopyObjectSSE(objectAPI ObjectLayer, instanceType, bucketNa // packages and catch a size that accounts for only one. large := bytes.Repeat([]byte("federated sse copy body "), 64*1024/24+1)[:64*1024+1] bodies := []struct { - name, ext string - data []byte + name, ext string + data []byte + requested, inherited hash.ChecksumType }{ + {name: "empty", ext: ".bin"}, + {name: "empty-explicit", ext: ".bin", requested: hash.ChecksumSHA256}, + {name: "inherited", ext: ".bin", data: []byte("stored encrypted checksum"), inherited: hash.ChecksumCRC32}, {name: "small", ext: ".bin", data: []byte("abc")}, {name: "two-packages", ext: ".bin", data: large}, {name: "compressed", ext: ".txt", data: large}, @@ -672,64 +807,74 @@ func testAPIFederatedCopyObjectSSE(objectAPI ObjectLayer, instanceType, bucketNa {"kms", "kms"}, {"plain", "kms-context"}, {"kms-context", "kms-context"}, + {"kms-context", "kms"}, + {"kms-context", "kms-empty-context"}, } const srcKeyByte, dstKeyByte = 0x11, 0x22 for _, body := range bodies { for _, pair := range pairs { + // Focus empty and inherited-checksum cases on each encrypted kind. + if body.name == "empty" || body.name == "empty-explicit" || body.name == "inherited" { + if pair.src != pair.dst || pair.src == "kms-context" { + continue + } + } t.Run(body.name+"/"+pair.src+"-to-"+pair.dst, func(t *testing.T) { prefix := "federation/sse-" + body.name + "-" + pair.src + "-to-" + pair.dst srcObject, dstObject := prefix+"-source"+body.ext, prefix+"-destination"+body.ext - putCopyChecksumSource(t, apiRouter, credentials, bucketName, srcObject, body.data, - federationSSEHeaders(pair.src, srcKeyByte, false)) + sourceHeaders := federationSSEHeaders(pair.src, srcKeyByte, false) + if pair.src == "kms-context" { + sourceHeaders[xhttp.AmzServerSideEncryptionKmsContext] = base64.StdEncoding.EncodeToString([]byte(`{"tenant":"source"}`)) + } + if body.inherited.IsSet() { + sourceHeaders[body.inherited.Key()] = mustChecksum(t, body.inherited, body.data) + } + putCopyChecksumSource(t, apiRouter, credentials, bucketName, srcObject, body.data, sourceHeaders) before, err := objectAPI.GetObjectInfo(t.Context(), bucketName, srcObject, ObjectOptions{}) if err != nil { t.Fatalf("%s: GetObjectInfo(source) failed: %v", instanceType, err) } - if got, want := federationStoredSSE(before.UserDefined), strings.TrimSuffix(pair.src, "-context"); got != want { + if got, want := federationStoredSSE(before.UserDefined), strings.Split(pair.src, "-")[0]; got != want { t.Fatalf("%s: source stored as %s, want %s", instanceType, got, want) } if compressed := body.ext == ".txt" && pair.src != "c"; before.IsCompressed() != compressed { t.Fatalf("%s: source compressed=%v, want %v", instanceType, before.IsCompressed(), compressed) } + assertFederationKMSContext(t, before, sourceHeaders[xhttp.AmzServerSideEncryptionKmsContext]) + if body.inherited.IsSet() { + sourceKeys := make(http.Header) + for k, v := range federationSSEHeaders(pair.src, srcKeyByte, true) { + sourceKeys.Set(k, v) + } + assertCopyChecksum(t, objectAPI, bucketName, srcObject, body.inherited, body.data, false, sourceKeys) + } headers := federationSSEHeaders(pair.dst, dstKeyByte, false) + if body.requested.IsSet() { + headers[xhttp.AmzChecksumAlgo] = body.requested.String() + } maps.Copy(headers, federationSSEHeaders(pair.src, srcKeyByte, true)) rec := federatedCopyRequest(t, apiRouter, credentials, bucketName, srcObject, remoteBucket, dstObject, headers) if rec.Code != http.StatusOK { t.Fatalf("%s: federated CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String()) } - // The remote computes the S3 default CRC-64NVME over the bytes it - // received, so the returned checksum must be the plaintext's. - assertCopyChecksumResponse(t, rec, hash.ChecksumCRC64NVME, body.data) - - // A destination GET must return the original plaintext at its length. + wantChecksum := hash.ChecksumCRC64NVME + if body.requested.IsSet() { + wantChecksum = body.requested + } else if body.inherited.IsSet() { + wantChecksum = body.inherited + } var getHeaders map[string]string if pair.dst == "c" { getHeaders = federationSSEHeaders("c", dstKeyByte, false) } - got := federationGetObject(t, apiRouter, credentials, remoteBucket, dstObject, getHeaders) - if got.Code != http.StatusOK { - t.Fatalf("%s: destination GET failed: %d %s", instanceType, got.Code, got.Body.String()) - } - if !bytes.Equal(got.Body.Bytes(), body.data) { - t.Fatalf("%s: destination GET returned %d bytes that differ from the %d-byte plaintext", - instanceType, got.Body.Len(), len(body.data)) - } - if want := strconv.Itoa(len(body.data)); got.Header().Get(xhttp.ContentLength) != want { - t.Fatalf("%s: destination Content-Length = %q, want %q", - instanceType, got.Header().Get(xhttp.ContentLength), want) - } - - // The destination is stored under the requested SSE kind and was - // encrypted exactly once: a second layer would add its own DARE - // package overhead to the stored size. - after, err := objectAPI.GetObjectInfo(t.Context(), remoteBucket, dstObject, ObjectOptions{}) - if err != nil { - t.Fatalf("%s: GetObjectInfo(destination) failed: %v", instanceType, err) - } - if got, want := federationStoredSSE(after.UserDefined), strings.TrimSuffix(pair.dst, "-context"); got != want { + after := assertFederationCopyReadback(t, objectAPI, apiRouter, credentials, remoteBucket, dstObject, + rec, wantChecksum, body.data, body.ext == ".txt" && pair.dst != "c", getHeaders) + assertFederationKMSContext(t, after, headers[xhttp.AmzServerSideEncryptionKmsContext]) + // A second encryption layer would add its own DARE overhead. + if got, want := federationStoredSSE(after.UserDefined), strings.Split(pair.dst, "-")[0]; got != want { t.Fatalf("%s: destination stored as %s, want %s", instanceType, got, want) } if pair.dst != "plain" && !after.IsCompressed() { @@ -753,46 +898,58 @@ func testAPIFederatedCopyObjectSSE(objectAPI ObjectLayer, instanceType, bucketNa } } - // A multipart SSE-S3 source is encrypted per part, so its logical size is - // the sum of the parts' decrypted sizes and the decrypting reader crosses - // a part boundary; the copy must still forward exactly the plaintext. + // Multipart sources cross an encryption boundary per part. SHA256 also + // exercises promotion from a stored composite to a full-object checksum. parts := [][]byte{bytes.Repeat([]byte("p"), globalMinPartSize+1), []byte("q!")} data := bytes.Join(parts, nil) - for _, dst := range []string{"plain", "s3"} { - t.Run("multipart/s3-to-"+dst, func(t *testing.T) { - srcObject, dstObject := "federation/sse-multipart-source-"+dst+".bin", "federation/sse-multipart-destination-"+dst+".bin" - putFederationMultipartSource(t, apiRouter, credentials, bucketName, srcObject, parts, - federationSSEHeaders("s3", srcKeyByte, false)) + for _, pair := range []struct { + src, dst string + composite bool + }{ + {src: "s3", dst: "plain"}, + {src: "s3", dst: "s3"}, + {src: "c", dst: "c"}, + {src: "kms", dst: "kms", composite: true}, + } { + t.Run("multipart/"+pair.src+"-to-"+pair.dst, func(t *testing.T) { + srcObject, dstObject := "federation/multipart-source-"+pair.src+"-"+pair.dst+".bin", "federation/multipart-destination-"+pair.src+"-"+pair.dst+".bin" + sourceHeaders := federationSSEHeaders(pair.src, srcKeyByte, false) + wantChecksum := hash.ChecksumCRC64NVME + if pair.composite { + wantChecksum = hash.ChecksumSHA256 + sourceHeaders[xhttp.AmzChecksumAlgo] = wantChecksum.String() + sourceHeaders[xhttp.AmzChecksumType] = xhttp.AmzChecksumTypeComposite + } + putFederationMultipartSource(t, apiRouter, credentials, bucketName, srcObject, parts, sourceHeaders) before, err := objectAPI.GetObjectInfo(t.Context(), bucketName, srcObject, ObjectOptions{}) if err != nil { - t.Fatalf("%s: GetObjectInfo(source) failed: %v", instanceType, err) + t.Fatal(err) } - if len(before.Parts) != len(parts) || federationStoredSSE(before.UserDefined) != "s3" { - t.Fatalf("%s: source has %d parts stored as %s, want %d parts as s3", - instanceType, len(before.Parts), federationStoredSSE(before.UserDefined), len(parts)) + if len(before.Parts) != len(parts) || federationStoredSSE(before.UserDefined) != pair.src { + t.Fatalf("source has %d parts stored as %s, want %d parts as %s", len(before.Parts), federationStoredSSE(before.UserDefined), len(parts), pair.src) } - - rec := federatedCopyRequest(t, apiRouter, credentials, bucketName, srcObject, remoteBucket, dstObject, - federationSSEHeaders(dst, dstKeyByte, false)) + if pair.composite { + checksums, _ := before.decryptChecksums(0, nil) + if checksums[xhttp.AmzChecksumType] != xhttp.AmzChecksumTypeComposite || !strings.HasSuffix(checksums[wantChecksum.String()], "-2") { + t.Fatalf("source did not persist a two-part composite checksum: %v", checksums) + } + } + headers := federationSSEHeaders(pair.dst, dstKeyByte, false) + maps.Copy(headers, federationSSEHeaders(pair.src, srcKeyByte, true)) + rec := federatedCopyRequest(t, apiRouter, credentials, bucketName, srcObject, remoteBucket, dstObject, headers) if rec.Code != http.StatusOK { t.Fatalf("%s: federated CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String()) } - assertCopyChecksumResponse(t, rec, hash.ChecksumCRC64NVME, data) - got := federationGetObject(t, apiRouter, credentials, remoteBucket, dstObject, nil) - if got.Code != http.StatusOK || !bytes.Equal(got.Body.Bytes(), data) { - t.Fatalf("%s: destination GET = %d with %d bytes, want 200 with the %d-byte plaintext", - instanceType, got.Code, got.Body.Len(), len(data)) + var getHeaders map[string]string + if pair.dst == "c" { + getHeaders = federationSSEHeaders("c", dstKeyByte, false) } - after, err := objectAPI.GetObjectInfo(t.Context(), remoteBucket, dstObject, ObjectOptions{}) - if err != nil { - t.Fatalf("%s: GetObjectInfo(destination) failed: %v", instanceType, err) + after := assertFederationCopyReadback(t, objectAPI, apiRouter, credentials, remoteBucket, dstObject, rec, wantChecksum, data, false, getHeaders) + if got := federationStoredSSE(after.UserDefined); got != pair.dst { + t.Fatalf("destination stored as %s, want %s", got, pair.dst) } - if got := federationStoredSSE(after.UserDefined); got != dst { - t.Fatalf("%s: destination stored as %s, want %s", instanceType, got, dst) - } - if once := (ObjectInfo{Size: int64(len(data))}); dst == "s3" && after.Size != once.EncryptedSize() { - t.Fatalf("%s: destination stored %d bytes, want %d for %d bytes encrypted once", - instanceType, after.Size, once.EncryptedSize(), len(data)) + if once := (ObjectInfo{Size: int64(len(data))}); pair.dst != "plain" && after.Size != once.EncryptedSize() { + t.Fatalf("destination stored %d bytes, want %d for %d bytes encrypted once", after.Size, once.EncryptedSize(), len(data)) } }) } diff --git a/cmd/object-federation-time_test.go b/cmd/object-federation-time_test.go new file mode 100644 index 000000000..f421cab15 --- /dev/null +++ b/cmd/object-federation-time_test.go @@ -0,0 +1,243 @@ +// Copyright (c) 2026 PGSTY +// SPDX-License-Identifier: AGPL-3.0-only + +package cmd + +import ( + "bytes" + "encoding/xml" + "fmt" + "io" + "maps" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + miniogo "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" + "github.com/minio/minio/internal/amztime" + "github.com/minio/minio/internal/auth" + xhttp "github.com/minio/minio/internal/http" + "github.com/minio/minio/internal/kms" +) + +// Exercise the selected SDK's actual retry loop and concurrent calls sharing a +// transport. Timestamps must follow the final PUT response, never a location +// probe, another operation, a failed attempt, or the HTTP Date header. +func TestFederatedWriteTimeTransport(t *testing.T) { + var mu sync.Mutex + attempts := make(map[string]int) + written := time.Date(2001, 2, 3, 4, 5, 6, 123456789, time.UTC) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + w.Header().Set(federatedLastModified, written.Add(-time.Hour).Format(time.RFC3339Nano)) + if r.Method == http.MethodGet && r.URL.Query().Has("location") { + _, _ = io.WriteString(w, `us-east-1`) + return + } + if r.Method != http.MethodPut { + t.Errorf("unexpected follow-up read: %s %s", r.Method, r.URL) + w.WriteHeader(http.StatusForbidden) + return + } + mu.Lock() + attempts[r.URL.Path]++ + attempt := attempts[r.URL.Path] + mu.Unlock() + if strings.Contains(r.URL.Path, "retry-") && attempt == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = io.WriteString(w, `SlowDown`) + return + } + stamp := written.Add(time.Duration(len(r.URL.Path)) * time.Nanosecond) + if r.URL.Query().Has("partNumber") { + stamp = stamp.Add(time.Second) + } + w.Header().Set(federatedLastModified, stamp.Format(time.RFC3339Nano)) + switch { + case strings.HasSuffix(r.URL.Path, "missing"): + w.Header().Del(federatedLastModified) + case strings.HasSuffix(r.URL.Path, "malformed"): + w.Header().Set(federatedLastModified, "invalid-time") + } + w.Header().Set("ETag", `"`+r.URL.Path+`"`) + })) + t.Cleanup(server.Close) + core, err := miniogo.NewCore(server.Listener.Addr().String(), &miniogo.Options{ + Creds: credentials.NewStaticV4("federation", "federation-secret", ""), + Transport: federatedWriteTransport{server.Client().Transport}, + BucketLookup: miniogo.BucketLookupPath, + MaxRetries: 2, + }) + if err != nil { + t.Fatal(err) + } + for _, part := range []bool{false, true} { + for _, scenario := range []string{"present", "missing", "malformed", "retry-present", "retry-missing", "retry-malformed"} { + t.Run(fmt.Sprintf("part=%v/%s", part, scenario), func(t *testing.T) { + t.Parallel() + ctx, modified := withFederatedWriteTime(t.Context()) + object := fmt.Sprintf("part-%v/%s", part, scenario) + var etag string + if part { + info, err := core.PutObjectPart(ctx, "bucket", object, "upload-id", 1, bytes.NewReader([]byte("abc")), 3, miniogo.PutObjectPartOptions{}) + if err != nil { + t.Fatal(err) + } + etag = info.ETag + } else { + info, err := core.PutObject(ctx, "bucket", object, bytes.NewReader([]byte("abc")), 3, "", "", miniogo.PutObjectOptions{}) + if err != nil { + t.Fatal(err) + } + etag = info.ETag + } + want := time.Time{} + if strings.HasSuffix(scenario, "present") { + want = written.Add(time.Duration(len("/bucket/"+object)) * time.Nanosecond) + if part { + want = want.Add(time.Second) + } + } + if !modified.Equal(want) || etag != "/bucket/"+object { + t.Errorf("write result: time=%s ETag=%s, want time=%s ETag=/bucket/%s", modified, etag, want, object) + } + mu.Lock() + count := attempts["/bucket/"+object] + mu.Unlock() + wantAttempts := 1 + if strings.HasPrefix(scenario, "retry-") { + wantAttempts = 2 + } + if count != wantAttempts { + t.Errorf("PUT attempts = %d, want %d", count, wantAttempts) + } + }) + } + } +} + +func TestAPIFederatedCopyWriteTimeSSEAndCompatibility(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ + t: t, + endpoints: []string{"CopyObjectPart", "NewMultipart", "PutObjectPart", "ListObjectParts", "CompleteMultipart", "CopyObject", "PutObject", "GetObject"}, + objAPITest: func(obj ObjectLayer, instanceType, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) { + testKMS, err := kms.NewBuiltin(federationTestKMSKeyID, bytes.Repeat([]byte{0x58}, 32)) + if err != nil { + t.Fatal(err) + } + previous := GlobalKMS + GlobalKMS = testKMS + defer func() { GlobalKMS = previous }() + for _, mode := range []string{"current", "missing", "malformed"} { + t.Run(mode, func(t *testing.T) { + remote, _, cleanup := setupCopyObjectFederationRemote(t, obj, router, instanceType, bucket, true, func(h http.Header) { + switch mode { + case "missing": + h.Del(federatedLastModified) + case "malformed": + h.Set(federatedLastModified, "invalid-time") + } + }) + defer cleanup() + for _, kind := range []string{"plain", "s3", "c", "kms"} { + if mode != "current" && kind != "plain" && kind != "c" { + continue + } + t.Run(kind, func(t *testing.T) { + data := []byte("federated write timestamp") + putCopyChecksumSource(t, router, cred, bucket, "source", data, federationSSEHeaders(kind, 0x11, false)) + headers := federationSSEHeaders(kind, 0x22, false) + maps.Copy(headers, federationSSEHeaders(kind, 0x11, true)) + rec := federatedCopyRequest(t, router, cred, bucket, "source", remote, kind+"-copy", headers) + if rec.Code != http.StatusOK { + t.Fatalf("CopyObject: %d %s", rec.Code, rec.Body.String()) + } + info, err := obj.GetObjectInfo(t.Context(), remote, kind+"-copy", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + var copied CopyObjectResponse + if err := xml.Unmarshal(rec.Body.Bytes(), &copied); err != nil { + t.Fatal(err) + } + assertTime := func(got string, stored time.Time) { + t.Helper() + if stored.IsZero() { + t.Fatal("storage returned a zero timestamp") + } + if mode != "current" { + stored = time.Time{} + } + if got != amztime.ISO8601Format(stored.UTC()) { + t.Errorf("copy time = %q, want %q", got, amztime.ISO8601Format(stored.UTC())) + } + } + assertTime(copied.LastModified, info.ModTime) + + object := kind + "-multipart" + req, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", remote, object), 0, nil, + cred.AccessKey, cred.SecretKey, federationSSEHeaders(kind, 0x22, false)) + if err != nil { + t.Fatal(err) + } + rec = httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("NewMultipartUpload: %d %s", rec.Code, rec.Body.String()) + } + var upload InitiateMultipartUploadResponse + if err := xml.Unmarshal(rec.Body.Bytes(), &upload); err != nil { + t.Fatal(err) + } + // UploadPartCopy selects KMS/S3 encryption from the upload; + // only SSE-C keys accompany the part request. The replica + // markers must still take the ordinary decrypting copy path. + headers = federationSSEHeaders(kind, 0x11, true) + var keys map[string]string + if kind == "c" { + keys = federationSSEHeaders("c", 0x22, false) + maps.Copy(headers, keys) + } + headers[xhttp.AmzCopySource] = SlashSeparator + pathJoin(bucket, "source") + headers[xhttp.MinIOSourceReplicationRequest] = "true" + headers[xhttp.AmzBucketReplicationStatus] = "REPLICA" + req, err = newTestSignedRequestV4(http.MethodPut, getCopyObjectPartURL("", remote, object, upload.UploadID, "1"), 0, nil, cred.AccessKey, cred.SecretKey, headers) + if err != nil { + t.Fatal(err) + } + rec = httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("UploadPartCopy: %d %s", rec.Code, rec.Body.String()) + } + var part CopyObjectPartResponse + if err := xml.Unmarshal(rec.Body.Bytes(), &part); err != nil { + t.Fatal(err) + } + parts, err := obj.ListObjectParts(t.Context(), remote, object, upload.UploadID, 0, 100, ObjectOptions{}) + if err != nil || len(parts.Parts) != 1 { + t.Fatalf("stored part: %v, %v", parts, err) + } + assertTime(part.LastModified, parts.Parts[0].LastModified) + rec = completePartsHTTP(t, router, cred, remote, object, upload.UploadID, []CompletePart{{PartNumber: 1, ETag: canonicalizeETag(part.ETag)}}, keys) + if rec.Code != http.StatusOK { + t.Fatalf("CompleteMultipartUpload: %d %s", rec.Code, rec.Body.String()) + } + for _, name := range []string{kind + "-copy", object} { + got := federationGetObject(t, router, cred, remote, name, keys) + if got.Code != http.StatusOK || !bytes.Equal(got.Body.Bytes(), data) { + t.Fatalf("readback %s: %d %s", name, got.Code, got.Body.String()) + } + } + }) + } + }) + } + }, + }) +} diff --git a/cmd/object-handlers-common.go b/cmd/object-handlers-common.go index f8de11ab8..438ba8d3a 100644 --- a/cmd/object-handlers-common.go +++ b/cmd/object-handlers-common.go @@ -36,6 +36,44 @@ import ( var etagRegex = regexp.MustCompile("\"*?([^\"]*?)\"*?$") +// federatedLastModified carries the committed object/part time on the write +// response. HTTP Date is a response time and Last-Modified loses subsecond +// precision, so neither can supply CopyObject/UploadPartCopy's timestamp. +const federatedLastModified = "X-Minio-Last-Modified" + +type federatedWriteTimeKey struct{} + +// withFederatedWriteTime allocates one result per SDK operation. SDK retries +// run sequentially with this context; concurrent copies never share the result. +func withFederatedWriteTime(ctx context.Context) (context.Context, *time.Time) { + modified := new(time.Time) + return context.WithValue(ctx, federatedWriteTimeKey{}, modified), modified +} + +type federatedWriteTransport struct { + http.RoundTripper +} + +func (t federatedWriteTransport) RoundTrip(r *http.Request) (*http.Response, error) { + resp, err := t.RoundTripper.RoundTrip(r) + if modified, ok := r.Context().Value(federatedWriteTimeKey{}).(*time.Time); ok && r.Method == http.MethodPut { + // Replace on every attempt, including a success from an older target + // without the header. Never retain a failed attempt's timestamp or fail + // a committed write just because its timestamp is unavailable. + *modified = time.Time{} + if err == nil && resp != nil && resp.StatusCode == http.StatusOK { + *modified, _ = time.Parse(time.RFC3339Nano, resp.Header.Get(federatedLastModified)) + } + } + return resp, err +} + +func setFederatedWriteTime(w http.ResponseWriter, r *http.Request, modified time.Time) { + if isFederatedInternalRequest(r.UserAgent()) && !modified.IsZero() { + w.Header().Set(federatedLastModified, modified.UTC().Format(time.RFC3339Nano)) + } +} + // Validates the preconditions for CopyObjectPart, returns true if CopyObjectPart // operation should not proceed. Preconditions supported are: // diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index 86321943b..c6262d2fd 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -1205,12 +1205,20 @@ const federatedInternalAppName = "minio-federated" // Applicable only in a federated deployment var getRemoteInstanceClient = func(r *http.Request, host string) (*miniogo.Core, error) { cred := getReqAccessCred(r, globalSite.Region()) + transport := getRemoteInstanceTransport() + if transport == nil { + var err error + transport, err = miniogo.DefaultTransport(globalIsTLS) + if err != nil { + return nil, err + } + } // In a federated deployment, all the instances share config files // and hence expected to have same credentials. core, err := miniogo.NewCore(host, &miniogo.Options{ Creds: credentials.NewStaticV4(cred.AccessKey, cred.SecretKey, ""), Secure: globalIsTLS, - Transport: getRemoteInstanceTransport(), + Transport: federatedWriteTransport{transport}, }) if err != nil { return nil, err @@ -1489,6 +1497,12 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re } defer gr.Close() srcInfo := gr.ObjInfo + // A trusted SSE-C replica reads raw ciphertext. Federation's ordinary PUT + // expects plaintext and does not forward the replica's sealed-key metadata. + if remoteCallRequired && replicaTrusted && crypto.SSEC.IsEncrypted(srcInfo.UserDefined) { + writeErrorResponse(ctx, w, toAPIError(ctx, NotImplemented{Message: "federated raw SSE-C replica CopyObject is not supported"}), r.URL) + return + } // maximum Upload size for object in a single CopyObject operation. if isMaxObjectSize(srcInfo.Size) { @@ -2007,10 +2021,11 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re if checksumHeaderValue != "" { opts.UserMetadata[wantChecksumType.Key()] = checksumHeaderValue } - // srcInfo.Reader yields the logical bytes, so declare the logical size: - // srcInfo.Size is the stored size, which differs for an encrypted or - // compressed source. - remoteObjInfo, rerr := core.PutObject(ctx, dstBucket, dstObject, srcInfo.Reader, + // For an ordinary federated copy, actualSize is the logical plaintext + // length to declare. srcInfo.Size is not a reliable wire length because + // the read path may already have adjusted it. + writeCtx, modified := withFederatedWriteTime(ctx) + remoteObjInfo, rerr := core.PutObject(writeCtx, dstBucket, dstObject, srcInfo.Reader, actualSize, "", "", opts) if rerr != nil { writeErrorResponse(ctx, w, toAPIError(ctx, rerr), r.URL) @@ -2029,7 +2044,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re objInfo.Size = actualSize objInfo.VersionID = remoteObjInfo.VersionID objInfo.ETag = remoteObjInfo.ETag - objInfo.ModTime = remoteObjInfo.LastModified + objInfo.ModTime = *modified // Bind the checksum the remote computed for this exact write. A single // forwarded PutObject must yield a full-object digest, so reject a // missing, malformed, or multipart-marked ("-N") value rather than @@ -2531,6 +2546,7 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req } setPutObjHeaders(w, objInfo, false, r.Header) + setFederatedWriteTime(w, r, objInfo.ModTime) // Notify object created event. evt := eventArgs{ diff --git a/cmd/object-multipart-federation-checksum_test.go b/cmd/object-multipart-federation-checksum_test.go index 85269c7ee..7b7707ece 100644 --- a/cmd/object-multipart-federation-checksum_test.go +++ b/cmd/object-multipart-federation-checksum_test.go @@ -30,6 +30,7 @@ import ( "strings" "sync" "testing" + "time" miniogo "github.com/minio/minio-go/v7" miniocredentials "github.com/minio/minio-go/v7/pkg/credentials" @@ -51,7 +52,7 @@ func TestAPIFederatedUploadPartChecksumResponse(t *testing.T) { }) } -func testAPIFederatedUploadPartChecksumResponse(_ ObjectLayer, instanceType, bucketName string, +func testAPIFederatedUploadPartChecksumResponse(objectAPI ObjectLayer, instanceType, bucketName string, apiRouter http.Handler, credentials auth.Credentials, t *testing.T, ) { algorithms := []struct { @@ -101,6 +102,18 @@ func testAPIFederatedUploadPartChecksumResponse(_ ObjectLayer, instanceType, buc if got := rec.Header().Get(xhttp.AmzChecksumType); got != "" { t.Fatalf("%s: UploadPart returned checksum type %q", instanceType, got) } + modified := rec.Header().Get(federatedLastModified) + if userAgent.want { + parts, err := objectAPI.ListObjectParts(t.Context(), bucketName, object, uploadID, 0, 100, ObjectOptions{}) + if err != nil || len(parts.Parts) != 1 { + t.Fatalf("stored part: %v, %v", parts, err) + } + if modified != parts.Parts[0].LastModified.UTC().Format(time.RFC3339Nano) { + t.Errorf("response time %q does not match committed part time %s", modified, parts.Parts[0].LastModified) + } + } else if modified != "" { + t.Errorf("ordinary UploadPart exposed federation timestamp %q", modified) + } }) } } @@ -158,7 +171,7 @@ func TestAPIFederatedUploadPartChecksumConcurrentOverwrite(t *testing.T) { }) } -func testAPIFederatedUploadPartChecksumConcurrentOverwrite(_ ObjectLayer, instanceType, bucketName string, +func testAPIFederatedUploadPartChecksumConcurrentOverwrite(objectAPI ObjectLayer, instanceType, bucketName string, apiRouter http.Handler, credentials auth.Credentials, t *testing.T, ) { object := "federation/concurrent-overwrite" @@ -194,6 +207,10 @@ func testAPIFederatedUploadPartChecksumConcurrentOverwrite(_ ObjectLayer, instan } close(start) wg.Wait() + parts, err := objectAPI.ListObjectParts(t.Context(), bucketName, object, uploadID, 0, 100, ObjectOptions{}) + if err != nil || len(parts.Parts) != 1 { + t.Fatalf("stored part: %v, %v", parts, err) + } for i, rec := range recorders { if rec.Code != http.StatusOK { @@ -213,6 +230,13 @@ func testAPIFederatedUploadPartChecksumConcurrentOverwrite(_ ObjectLayer, instan if want := hex.EncodeToString(md5sum[:]); canonicalizeETag(etags[0]) != want { t.Fatalf("%s: concurrent request %d ETag %q, want %q", instanceType, i, etags[0], want) } + modified, err := time.Parse(time.RFC3339Nano, rec.Header().Get(federatedLastModified)) + if err != nil || modified.IsZero() { + t.Fatalf("concurrent request %d returned invalid time %v: %v", i, modified, err) + } + if canonicalizeETag(etags[0]) == parts.Parts[0].ETag && !modified.Equal(parts.Parts[0].LastModified) { + t.Errorf("surviving part has time %s, its writer returned %s", parts.Parts[0].LastModified, modified) + } } } @@ -352,6 +376,9 @@ func testAPIFederatedCopyObjectPartChecksum(objectAPI ObjectLayer, instanceType, if len(parts.Parts) != 1 { t.Fatalf("%s: ListParts returned %d parts, want 1", instanceType, len(parts.Parts)) } + if response.LastModified != parts.Parts[0].LastModified { + t.Errorf("CopyPartResult time = %q, want stored part time %q", response.LastModified, parts.Parts[0].LastModified) + } if got := partChecksum(algorithm.typ, parts.Parts[0]); got != want { t.Fatalf("%s: persisted part %s is %q, want %q", instanceType, algorithm.typ.String(), got, want) } diff --git a/cmd/object-multipart-handlers.go b/cmd/object-multipart-handlers.go index 983644223..44f61f353 100644 --- a/cmd/object-multipart-handlers.go +++ b/cmd/object-multipart-handlers.go @@ -59,8 +59,8 @@ import ( // // This is only a response-shape hint. User-Agent is not authenticated and must // never gate authorization, object visibility, or request validation. It is -// safe here because the only effect is returning the checksum of the body the -// caller was already authorized to upload. +// safe here because the only effect is returning the checksum and modification +// time of the body the caller was already authorized to upload. func isFederatedInternalRequest(userAgent string) bool { for _, product := range strings.Fields(userAgent) { name, version, ok := strings.Cut(product, "/") @@ -572,7 +572,8 @@ func (api objectAPIHandlers) CopyObjectPartHandler(w http.ResponseWriter, r *htt SSE: dstOpts.ServerSideEncryption, } - partInfo, err := core.PutObjectPart(ctx, dstBucket, dstObject, uploadID, partID, gr, length, popts) + writeCtx, modified := withFederatedWriteTime(ctx) + partInfo, err := core.PutObjectPart(writeCtx, dstBucket, dstObject, uploadID, partID, gr, length, popts) if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) return @@ -580,7 +581,7 @@ func (api objectAPIHandlers) CopyObjectPartHandler(w http.ResponseWriter, r *htt response := generateCopyObjectPartResponse(PartInfo{ ETag: partInfo.ETag, - LastModified: partInfo.LastModified, + LastModified: *modified, ChecksumCRC32: partInfo.ChecksumCRC32, ChecksumCRC32C: partInfo.ChecksumCRC32C, ChecksumSHA1: partInfo.ChecksumSHA1, @@ -1082,6 +1083,7 @@ func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http // checksum cannot be mixed with a concurrent overwrite. hash.AddChecksumHeader(w, partChecksumMap(partInfo)) } + setFederatedWriteTime(w, r, partInfo.LastModified) writeSuccessResponseHeadersOnly(w) } diff --git a/docs/federation/lookup/README.md b/docs/federation/lookup/README.md index ef6fc8bdb..d76497234 100644 --- a/docs/federation/lookup/README.md +++ b/docs/federation/lookup/README.md @@ -2,6 +2,32 @@ This document explains how to configure Silo with `Bucket lookup from DNS` style federation. +## Cross-deployment copy behavior + +Federated `CopyObject` and `UploadPartCopy` forward the write to the destination +deployment. Updated Silo destinations return the modification time of that +committed object or part in `X-Minio-Last-Modified` (UTC RFC 3339 with nanosecond +precision), alongside its ETag and checksums. The proxy uses that write response +to populate the copy result, whose XML retains millisecond precision. This works +for unversioned and versioned objects and encrypted writes, without an additional +HEAD or ListParts request or additional read permissions at the destination. + +Both deployments need this change for accurate copy timestamps. If an older +destination omits the header, or an intermediary removes or corrupts it, the copy +still succeeds and retains the historical zero `LastModified` value +(`0001-01-01T00:00:00.000Z`). The proxy neither substitutes its own clock or HTTP +`Date` nor reports a committed write as failed because its timestamp is missing. +The header is a response hint for the existing federation User-Agent token; it +does not grant replication privileges. + +A replica-authorized `CopyObject` of a raw SSE-C source across deployments is +unsupported and returns HTTP 501 `NotImplemented` before forwarding any request. +This includes empty objects, which previously succeeded accidentally, and applies +whether or not a copy-source key is supplied. Ordinary federated SSE-C copies +with the source key, local SSE-C key rotation, bucket replication via PUT or +multipart upload, and the separate `UploadPartCopy` operation retain their +existing behavior. + ## Get started ### 1. Prerequisites