diff --git a/cmd/erasure-server-pool-recovery_test.go b/cmd/erasure-server-pool-recovery_test.go index 4ea38fd38..bf1159e72 100644 --- a/cmd/erasure-server-pool-recovery_test.go +++ b/cmd/erasure-server-pool-recovery_test.go @@ -20,13 +20,18 @@ package cmd import ( "bytes" "context" + "crypto/md5" "crypto/sha256" + "encoding/base64" "io" "net/http" "net/http/httptest" "sync/atomic" "testing" "time" + + xhttp "github.com/minio/minio/internal/http" + "github.com/minio/minio/internal/kms" ) // An acknowledged recovery overwrite must resolve the unreadable pool, even @@ -34,74 +39,103 @@ import ( // Real inline metadata is combined in the same 15-old/1-new state used by // that regression. Capacity adapters only make destination choice deterministic. func TestPoolsMinorityOverwriteRecovery(t *testing.T) { - for _, mode := range []string{"single-pool", "dual-pool-only-inconsistent-copy", "dual-pool-readable-other-copy"} { - t.Run(mode, func(t *testing.T) { - var z *erasureServerPools - if mode == "single-pool" { - z, _ = mutationPools(t) - } else { - z, _ = consistencyPools(t) - } - const object = "review-retry" - for _, pool := range z.serverPools { - restore := conditionalPutSwapDisks(pool, object, func(d StorageAPI) StorageAPI { return conditionalPutCapacityDisk{StorageAPI: d} }) - defer restore() - } - bucket, router := conditionalPutBucket(t, z, "unversioned") - oldTime := time.Now().UTC().Add(-time.Hour) - oldBody := bytes.Repeat([]byte{'a'}, 8192) - pendingBody := bytes.Repeat([]byte{'b'}, 8192) - recoveryBody := bytes.Repeat([]byte{'c'}, 8192) - seed := func(pool int, body []byte, when time.Time) { - t.Helper() - putConsistencyObject(t, z, bucket, object, pool, string(body), ObjectOptions{MTime: when}) - } - seed(0, oldBody, oldTime) - disks := z.serverPools[0].getHashedSet(object).getDisks() - oldMeta := make([][]byte, len(disks)) - for i, disk := range disks { - oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object) - } - seed(0, pendingBody, oldTime.Add(time.Minute)) - for i := 0; i < len(disks)-1; i++ { - if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil { - t.Fatal(err) - } - } - if mode == "dual-pool-readable-other-copy" { - seed(1, oldBody, oldTime.Add(-time.Minute)) - // Ordinary allocation can select this healthy other pool. - // Force that permitted choice without altering metadata reads. - restore := conditionalPutPool(t, z, object, 1) - defer restore() - } - before := consistencyRequest(t, router, http.MethodGet, bucket, object, "") - if before.Code != http.StatusServiceUnavailable { - t.Fatalf("fixture must refuse ambiguous read; got %d", before.Code) - } - req, err := newTestSignedRequestV4(http.MethodPut, getGetObjectURL("", bucket, object), int64(len(recoveryBody)), bytes.NewReader(recoveryBody), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) - if err != nil { - t.Fatal(err) - } - result := httptest.NewRecorder() - router.ServeHTTP(result, req.WithContext(t.Context())) - t.Logf("recovery PUT HTTP=%d response=%s", result.Code, result.Body.String()) - if result.Code != http.StatusOK { - t.Fatalf("unconditional recovery PUT: %d %s", result.Code, result.Body.String()) - } - after := consistencyRequest(t, router, http.MethodGet, bucket, object, "") - t.Logf("after acknowledged recovery PUT: GET HTTP=%d matches_new=%t", after.Code, bytes.Equal(after.Body.Bytes(), recoveryBody)) - for i, pool := range z.serverPools { - oi, err := pool.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}) - t.Logf("pool=%d mtime=%s err=%v", i, oi.ModTime, err) - } - if after.Code != http.StatusOK || !bytes.Equal(after.Body.Bytes(), recoveryBody) { - t.Fatalf("acknowledged recovery overwrite did not restore reads: HTTP=%d body=%s", after.Code, after.Body.String()) - } + for _, versioning := range []string{"unversioned", "versioned", "suspended"} { + for _, mode := range []string{"single-pool", "dual-pool-only-inconsistent-copy", "dual-pool-readable-other-copy"} { + t.Run(versioning+"/"+mode, func(t *testing.T) { + testPoolsMinorityOverwriteRecovery(t, mode, versioning, "") + }) + } + } + for _, encryption := range []string{"SSE-C", "SSE-S3", "SSE-KMS"} { + t.Run(encryption, func(t *testing.T) { + testPoolsMinorityOverwriteRecovery(t, "dual-pool-readable-other-copy", "unversioned", encryption) }) } } +func testPoolsMinorityOverwriteRecovery(t *testing.T, mode, versioning, encryption string) { + t.Helper() + var z *erasureServerPools + if mode == "single-pool" { + z, _ = mutationPools(t) + } else { + z, _ = consistencyPools(t) + } + const object = "review-retry" + for _, pool := range z.serverPools { + restore := conditionalPutSwapDisks(pool, object, func(d StorageAPI) StorageAPI { return conditionalPutCapacityDisk{StorageAPI: d} }) + defer restore() + } + bucket, router := conditionalPutBucket(t, z, versioning) + oldTime := time.Now().UTC().Add(-time.Hour) + oldBody := bytes.Repeat([]byte{'a'}, 8192) + pendingBody := bytes.Repeat([]byte{'b'}, 8192) + recoveryBody := bytes.Repeat([]byte{'c'}, 8192) + seed := func(pool int, body []byte, when time.Time) { + t.Helper() + putConsistencyObject(t, z, bucket, object, pool, string(body), ObjectOptions{MTime: when}) + } + seed(0, oldBody, oldTime) + disks := z.serverPools[0].getHashedSet(object).getDisks() + oldMeta := make([][]byte, len(disks)) + for i, disk := range disks { + oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object) + } + seed(0, pendingBody, oldTime.Add(time.Minute)) + for i := 0; i < len(disks)-1; i++ { + if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil { + t.Fatal(err) + } + } + if mode == "dual-pool-readable-other-copy" { + seed(1, oldBody, oldTime.Add(-time.Minute)) + // Ordinary allocation can select this healthy other pool. + // Force that permitted choice without altering metadata reads. + restore := conditionalPutPool(t, z, object, 1) + defer restore() + } + before := consistencyRequest(t, router, http.MethodGet, bucket, object, "") + if before.Code != http.StatusServiceUnavailable { + t.Fatalf("fixture must refuse ambiguous read; got %d", before.Code) + } + var headers, readHeaders map[string]string + if encryption != "" { + oldKMS, oldTLS := GlobalKMS, globalIsTLS + GlobalKMS, globalIsTLS = kms.NewStub("recovery-key"), true + defer func() { GlobalKMS, globalIsTLS = oldKMS, oldTLS }() + headers = map[string]string{xhttp.AmzServerSideEncryption: xhttp.AmzEncryptionAES} + switch encryption { + case "SSE-C": + key := bytes.Repeat([]byte{0x42}, 32) + digest := md5.Sum(key) + headers = map[string]string{ + xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES, + xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key), + xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(digest[:]), + } + readHeaders = headers + case "SSE-KMS": + headers[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionKMS + headers[xhttp.AmzServerSideEncryptionKmsID] = "recovery-key" + } + } + url := getPutObjectURL("", bucket, object) + result := multipartConditionRequest(t, router, http.MethodPut, url, string(recoveryBody), headers) + t.Logf("recovery PUT HTTP=%d response=%s", result.Code, result.Body.String()) + if result.Code != http.StatusOK { + t.Fatalf("unconditional recovery PUT: %d %s", result.Code, result.Body.String()) + } + after := multipartConditionRequest(t, router, http.MethodGet, url, "", readHeaders) + t.Logf("after acknowledged recovery PUT: GET HTTP=%d matches_new=%t", after.Code, bytes.Equal(after.Body.Bytes(), recoveryBody)) + for i, pool := range z.serverPools { + oi, err := pool.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}) + t.Logf("pool=%d mtime=%s err=%v", i, oi.ModTime, err) + } + if after.Code != http.StatusOK || !bytes.Equal(after.Body.Bytes(), recoveryBody) { + t.Fatalf("acknowledged recovery overwrite did not restore reads: HTTP=%d body=%s", after.Code, after.Body.String()) + } +} + // Read failures activated by a completed disk mutation verify that a recovery // ACK is not issued from preflight information alone. type recoveryReadFaultDisk struct {