Merge pull request #127 from pgsty/fix/issue-112-bucket-metadata-cors

fix: include per-bucket CORS in bucket metadata export and import

cmd/admin-bucket-cors-roundtrip_test.go

@@ -0,0 +1,348 @@
+package cmd
+
+import (
+	"archive/zip"
+	"bytes"
+	"encoding/base64"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/auth"
+	"github.com/minio/mux"
+)
+
+func corsAdminRequest(t *testing.T, cred auth.Credentials, method, path string, body []byte) *httptest.ResponseRecorder {
+	t.Helper()
+	router := mux.NewRouter()
+	registerAdminRouter(router, true)
+	req, err := newTestSignedRequestV4(method, adminPathPrefix+adminAPIVersionPrefix+path,
+		int64(len(body)), bytes.NewReader(body), cred.AccessKey, cred.SecretKey, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("admin %s: %d: %s", path, rec.Code, rec.Body.String())
+	}
+	return rec
+}
+
+func corsImportReport(t *testing.T, rec *httptest.ResponseRecorder) madmin.BucketMetaImportErrs {
+	t.Helper()
+	var rpt madmin.BucketMetaImportErrs
+	if err := json.Unmarshal(rec.Body.Bytes(), &rpt); err != nil {
+		t.Fatalf("import report %q: %v", rec.Body.String(), err)
+	}
+	return rpt
+}
+
+func corsZip(t *testing.T, entries map[string][]byte) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	zw := zip.NewWriter(&buf)
+	for name, data := range entries {
+		w, err := zw.Create(name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err = w.Write(data); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := zw.Close(); err != nil {
+		t.Fatal(err)
+	}
+	return buf.Bytes()
+}
+
+// corsCorruptedZip builds an archive holding a stored (uncompressed) cors.xml
+// whose payload is altered after the checksum is computed, plus the given
+// companion entries. The altered document stays well formed, so only the zip
+// checksum tells the two apart.
+func corsCorruptedZip(t *testing.T, name string, doc []byte, others map[string][]byte) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	zw := zip.NewWriter(&buf)
+	w, err := zw.CreateHeader(&zip.FileHeader{Name: name, Method: zip.Store})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err = w.Write(doc); err != nil {
+		t.Fatal(err)
+	}
+	for other, data := range others {
+		ow, err := zw.Create(other)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err = ow.Write(data); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err = zw.Close(); err != nil {
+		t.Fatal(err)
+	}
+	raw := buf.Bytes()
+	at := bytes.Index(raw, []byte("app.example.com"))
+	if at < 0 {
+		t.Fatalf("stored CORS payload not found in archive")
+	}
+	raw[at] = 'A'
+	return raw
+}
+
+// TestAdminBucketMetadataCORSRoundTrip covers the export/import round trip for
+// per-bucket CORS, per-file error reporting for an invalid document, and that
+// an archive without cors.xml leaves an existing configuration alone.
+func TestAdminBucketMetadataCORSRoundTrip(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: func(obj ObjectLayer, instanceType, bucket string, _ http.Handler, cred auth.Credentials, t *testing.T) {
+		corsXML := []byte(testSiteReplicationCORSDoc)
+		if _, err := updateLocalBucketCORSMetadata(t.Context(), obj, bucket, corsXML); err != nil {
+			t.Fatal(err)
+		}
+
+		// Export must carry the stored document verbatim.
+		rec := corsAdminRequest(t, cred, http.MethodGet, "/export-bucket-metadata?bucket="+bucket, nil)
+		archive := rec.Body.Bytes()
+		zr, err := zip.NewReader(bytes.NewReader(archive), int64(len(archive)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		var exported []byte
+		for _, f := range zr.File {
+			if f.Name != bucket+"/"+bucketCorsConfig {
+				continue
+			}
+			r, err := f.Open()
+			if err != nil {
+				t.Fatal(err)
+			}
+			exported, err = io.ReadAll(r)
+			r.Close()
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+		if !bytes.Equal(exported, corsXML) {
+			t.Fatalf("%s: exported CORS = %q, want %q", instanceType, exported, corsXML)
+		}
+
+		// Drop the configuration: the archive must then omit the entry.
+		if _, err = updateLocalBucketCORSMetadata(t.Context(), obj, bucket, nil); err != nil {
+			t.Fatal(err)
+		}
+		if _, _, err = globalBucketMetadataSys.GetCorsConfigXML(bucket); err == nil {
+			t.Fatalf("%s: CORS still present before restore", instanceType)
+		}
+		rec = corsAdminRequest(t, cred, http.MethodGet, "/export-bucket-metadata?bucket="+bucket, nil)
+		empty := rec.Body.Bytes()
+		zr, err = zip.NewReader(bytes.NewReader(empty), int64(len(empty)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, f := range zr.File {
+			if f.Name == bucket+"/"+bucketCorsConfig {
+				t.Fatalf("%s: export emitted %s for a bucket without CORS", instanceType, f.Name)
+			}
+		}
+
+		rec = corsAdminRequest(t, cred, http.MethodPut, "/import-bucket-metadata", archive)
+		if st := corsImportReport(t, rec).Buckets[bucket]; !st.Cors.IsSet || st.Cors.Err != "" {
+			t.Fatalf("%s: import report cors = %+v", instanceType, st.Cors)
+		}
+		stored, storedAt, err := globalBucketMetadataSys.GetCorsConfigXML(bucket)
+		if err != nil || !bytes.Equal(stored, corsXML) {
+			t.Fatalf("%s: restored CORS = %q, err = %v", instanceType, stored, err)
+		}
+		created, err := globalBucketMetadataSys.CreatedAt(bucket)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !storedAt.After(created) {
+			t.Fatalf("%s: restored CORS timestamp %v is not after bucket creation %v", instanceType, storedAt, created)
+		}
+
+		// An archive without cors.xml must not remove the configuration.
+		corsAdminRequest(t, cred, http.MethodPut, "/import-bucket-metadata",
+			corsZip(t, map[string][]byte{bucket + "/quota.json": []byte(`{"quota":0}`)}))
+		if stored, _, err = globalBucketMetadataSys.GetCorsConfigXML(bucket); err != nil || !bytes.Equal(stored, corsXML) {
+			t.Fatalf("%s: import without cors.xml changed CORS: %q, err = %v", instanceType, stored, err)
+		}
+
+		// A bucket the import itself creates must still land above its own
+		// creation time, otherwise CORS replication would drop the restore.
+		fresh := "cors-import-created-bucket"
+		rec = corsAdminRequest(t, cred, http.MethodPut, "/import-bucket-metadata",
+			corsZip(t, map[string][]byte{fresh + "/" + bucketCorsConfig: corsXML}))
+		if st := corsImportReport(t, rec).Buckets[fresh]; !st.Cors.IsSet || st.Cors.Err != "" {
+			t.Fatalf("%s: fresh bucket import report cors = %+v", instanceType, st.Cors)
+		}
+		freshStored, freshAt, err := globalBucketMetadataSys.GetCorsConfigXML(fresh)
+		if err != nil || !bytes.Equal(freshStored, corsXML) {
+			t.Fatalf("%s: fresh bucket CORS = %q, err = %v", instanceType, freshStored, err)
+		}
+		freshCreated, err := globalBucketMetadataSys.CreatedAt(fresh)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !freshAt.After(freshCreated) {
+			t.Fatalf("%s: fresh bucket CORS timestamp %v is not after creation %v", instanceType, freshAt, freshCreated)
+		}
+
+		// An invalid document must fail loudly for that bucket and change nothing.
+		rec = corsAdminRequest(t, cred, http.MethodPut, "/import-bucket-metadata",
+			corsZip(t, map[string][]byte{bucket + "/" + bucketCorsConfig: []byte("<CORSConfiguration><CORSRule>")}))
+		if st := corsImportReport(t, rec).Buckets[bucket]; st.Cors.Err == "" {
+			t.Fatalf("%s: invalid CORS import reported no error: %+v", instanceType, st)
+		}
+		if stored, _, err = globalBucketMetadataSys.GetCorsConfigXML(bucket); err != nil || !bytes.Equal(stored, corsXML) {
+			t.Fatalf("%s: invalid CORS import changed stored config: %q, err = %v", instanceType, stored, err)
+		}
+
+		// A well formed document carried by a corrupt zip entry must be
+		// rejected too, leaving the stored document and its timestamp alone
+		// while the other configs in the same archive still apply.
+		_, corsAt, err := globalBucketMetadataSys.GetCorsConfigXML(bucket)
+		if err != nil {
+			t.Fatal(err)
+		}
+		rec = corsAdminRequest(t, cred, http.MethodPut, "/import-bucket-metadata",
+			corsCorruptedZip(t, bucket+"/"+bucketCorsConfig, corsXML,
+				map[string][]byte{bucket + "/quota.json": []byte(`{"quota":4096,"quotatype":"hard"}`)}))
+		st := corsImportReport(t, rec).Buckets[bucket]
+		if st.Cors.Err == "" {
+			t.Fatalf("%s: corrupt CORS entry reported no error: %+v", instanceType, st)
+		}
+		if !st.Quota.IsSet || st.Quota.Err != "" {
+			t.Fatalf("%s: corrupt CORS entry blocked the neighboring quota: %+v", instanceType, st.Quota)
+		}
+		stored, storedAt, err = globalBucketMetadataSys.GetCorsConfigXML(bucket)
+		if err != nil || !bytes.Equal(stored, corsXML) || !storedAt.Equal(corsAt) {
+			t.Fatalf("%s: corrupt CORS entry changed stored config: %q at %v (was %v), err = %v", instanceType, stored, storedAt, corsAt, err)
+		}
+		quota, _, err := globalBucketMetadataSys.GetQuotaConfig(t.Context(), bucket)
+		if err != nil || quota == nil || quota.Quota != 4096 {
+			t.Fatalf("%s: neighboring quota not applied: %+v, err = %v", instanceType, quota, err)
+		}
+	}})
+}
+
+// corsPeerStub is a stand-in site-replication peer. It records every
+// SRBucketMeta it is asked to apply and answers with status.
+func corsPeerStub(t *testing.T, applied chan<- madmin.SRBucketMeta, status int) *httptest.Server {
+	t.Helper()
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodPut && applied != nil {
+			var item madmin.SRBucketMeta
+			if err := json.NewDecoder(r.Body).Decode(&item); err != nil {
+				t.Errorf("decode peer apply: %v", err)
+				w.WriteHeader(http.StatusBadRequest)
+				return
+			}
+			applied <- item
+		}
+		w.WriteHeader(status)
+	}))
+}
+
+// TestAdminBucketMetadataCORSImportReplicatesPastPeerFailure pins that an
+// imported CORS document reaches the reachable peers even when the shared
+// bucket metadata hook failed against an unreachable one, and that both
+// failures are still reported for the bucket.
+func TestAdminBucketMetadataCORSImportReplicatesPastPeerFailure(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: func(obj ObjectLayer, instanceType, bucket string, _ http.Handler, cred auth.Credentials, t *testing.T) {
+		ctx := t.Context()
+		corsXML := []byte(testSiteReplicationCORSDoc)
+
+		healthyApplies := make(chan madmin.SRBucketMeta, 4)
+		healthy := corsPeerStub(t, healthyApplies, http.StatusOK)
+		defer healthy.Close()
+		broken := corsPeerStub(t, nil, http.StatusBadRequest)
+		defer broken.Close()
+
+		// With site replication on, admin requests resolve their token signing
+		// key through the site replicator account, so it has to exist.
+		serviceCred, err := auth.CreateCredentials(siteReplicatorSvcAcc, "cors-import-service-secret")
+		if err != nil {
+			t.Fatal(err)
+		}
+		serviceCred.ParentUser = cred.AccessKey
+		if _, err = globalIAMSys.store.AddServiceAccount(ctx, serviceCred); err != nil {
+			t.Fatal(err)
+		}
+		defer globalIAMSys.DeleteServiceAccount(ctx, serviceCred.AccessKey, false)
+		globalSiteReplicatorCred.Set(serviceCred.SecretKey)
+		defer globalSiteReplicatorCred.Set("")
+
+		globalSiteReplicationSys.Lock()
+		oldEnabled, oldState := globalSiteReplicationSys.enabled, globalSiteReplicationSys.state
+		globalSiteReplicationSys.enabled = true
+		globalSiteReplicationSys.state = srState{
+			Name:                    "cors-import-test",
+			ServiceAccountAccessKey: serviceCred.AccessKey,
+			Peers: map[string]madmin.PeerInfo{
+				globalDeploymentID(): {Name: "local", DeploymentID: globalDeploymentID()},
+				"peer-healthy":       {Name: "healthy", DeploymentID: "peer-healthy", Endpoint: healthy.URL},
+				"peer-broken":        {Name: "broken", DeploymentID: "peer-broken", Endpoint: broken.URL},
+			},
+		}
+		globalSiteReplicationSys.Unlock()
+		defer func() {
+			globalSiteReplicationSys.Lock()
+			globalSiteReplicationSys.enabled, globalSiteReplicationSys.state = oldEnabled, oldState
+			globalSiteReplicationSys.Unlock()
+		}()
+
+		rec := corsAdminRequest(t, cred, http.MethodPut, "/import-bucket-metadata",
+			corsZip(t, map[string][]byte{
+				bucket + "/" + bucketCorsConfig: corsXML,
+				bucket + "/quota.json":          []byte(`{"quota":8192,"quotatype":"hard"}`),
+			}))
+		st := corsImportReport(t, rec).Buckets[bucket]
+		if !st.Cors.IsSet || st.Cors.Err != "" {
+			t.Fatalf("%s: import report cors = %+v", instanceType, st.Cors)
+		}
+		stored, storedAt, err := globalBucketMetadataSys.GetCorsConfigXML(bucket)
+		if err != nil || !bytes.Equal(stored, corsXML) {
+			t.Fatalf("%s: stored CORS = %q, err = %v", instanceType, stored, err)
+		}
+
+		// The reachable peer must have been told about the CORS document,
+		// carrying exactly the timestamp that was saved locally.
+		var corsSeen, sharedSeen bool
+		for range 2 {
+			select {
+			case item := <-healthyApplies:
+				if item.Type != madmin.SRBucketMetaTypeCorsConfig {
+					sharedSeen = item.Bucket == bucket && item.Quota != nil
+					continue
+				}
+				if item.Bucket != bucket || item.Cors == nil || !item.UpdatedAt.Equal(storedAt) {
+					t.Fatalf("%s: peer CORS event = %#v, want %s at %v", instanceType, item, bucket, storedAt)
+				}
+				payload, decErr := base64.StdEncoding.Strict().DecodeString(*item.Cors)
+				if decErr != nil || !bytes.Equal(payload, corsXML) {
+					t.Fatalf("%s: peer CORS payload = %q, err = %v", instanceType, payload, decErr)
+				}
+				corsSeen = true
+			case <-time.After(10 * time.Second):
+				t.Fatalf("%s: healthy peer received no further events (shared=%v cors=%v)", instanceType, sharedSeen, corsSeen)
+			}
+		}
+		if !sharedSeen || !corsSeen {
+			t.Fatalf("%s: healthy peer events shared=%v cors=%v, want both", instanceType, sharedSeen, corsSeen)
+		}
+
+		// Both hook failures against the unreachable peer stay reported.
+		if got := strings.Count(st.Err, "->broken:"); got != 2 {
+			t.Fatalf("%s: bucket error mentions the broken peer %d times, want 2: %q", instanceType, got, st.Err)
+		}
+	}})
+}

cmd/admin-bucket-handlers.go

@@ -417,6 +417,7 @@ func (a adminAPIHandlers) ExportBucketMetadataHandler(w http.ResponseWriter, r *
 		bucketLifecycleConfig,
 		bucketSSEConfig,
 		bucketTaggingConfig,
+		bucketCorsConfig,
 		bucketQuotaConfigFile,
 		objectLockConfig,
 		bucketVersioningConfig,
@@ -517,6 +518,19 @@ func (a adminAPIHandlers) ExportBucketMetadataHandler(w http.ResponseWriter, r *
 					return
 				}
 				rawDataFn(bytes.NewReader(configData), cfgPath, len(configData))
+			case bucketCorsConfig:
+				// Export the stored document verbatim: GetBucketCors returns
+				// the bytes exactly as they were PUT, so the archive must
+				// round-trip them unchanged.
+				configData, _, err := globalBucketMetadataSys.GetCorsConfigXML(bucket)
+				if err != nil {
+					if errors.Is(err, errConfigNotFound) {
+						continue
+					}
+					writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
+					return
+				}
+				rawDataFn(bytes.NewReader(configData), cfgPath, len(configData))
 			case objectLockConfig:
 				config, _, err := globalBucketMetadataSys.GetObjectLockConfig(bucket)
 				if err != nil {
@@ -616,6 +630,13 @@ func applyImportedBucketMetadata(dst *BucketMetadata, src BucketMetadata, fields
 		case bucketQuotaConfigFile:
 			dst.QuotaConfigJSON = bytes.Clone(src.QuotaConfigJSON)
 			dst.QuotaConfigUpdatedAt = src.QuotaConfigUpdatedAt
+		case bucketCorsConfig:
+			// The import stamps its fields before creating any missing bucket,
+			// and a CORS event stamped before bucket creation is discarded as
+			// belonging to an older incarnation, so the imported document takes
+			// the same monotonic timestamp a local PutBucketCors would assign.
+			dst.CorsConfigUpdatedAt = localCORSUpdatedAt(*dst, src.CorsConfigUpdatedAt)
+			dst.CorsConfigXML = bytes.Clone(src.CorsConfigXML)
 		case objectLockConfig:
 			dst.ObjectLockConfigXML = bytes.Clone(src.ObjectLockConfigXML)
 			dst.ObjectLockConfigUpdatedAt = src.ObjectLockConfigUpdatedAt
@@ -645,6 +666,8 @@ func (i *importMetaReport) SetStatus(bucket, fname string, err error) {
 		st.Tagging = madmin.MetaStatus{IsSet: true, Err: errMsg}
 	case bucketQuotaConfigFile:
 		st.Quota = madmin.MetaStatus{IsSet: true, Err: errMsg}
+	case bucketCorsConfig:
+		st.Cors = madmin.MetaStatus{IsSet: true, Err: errMsg}
 	case objectLockConfig:
 		st.ObjectLock = madmin.MetaStatus{IsSet: true, Err: errMsg}
 	case bucketVersioningConfig:
@@ -1015,6 +1038,32 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 			bucketMap[bucket].QuotaConfigUpdatedAt = updatedAt
 			markImported(bucket, fileName)
 			rpt.SetStatus(bucket, fileName, nil)
+		case bucketCorsConfig:
+			if sz > maxBucketCorsSize {
+				rpt.SetStatus(bucket, fileName, errors.New(ErrEntityTooLarge.String()))
+				continue
+			}
+
+			// Read one byte past the declared size: stopping exactly at sz
+			// leaves archive/zip short of EOF, so it never verifies the entry
+			// checksum and a corrupt entry carrying well formed XML would be
+			// stored as a valid document. The extra byte also lets the reader
+			// reject an entry longer than it declares.
+			corsData, err := io.ReadAll(io.LimitReader(reader, sz+1))
+			if err != nil {
+				rpt.SetStatus(bucket, fileName, err)
+				continue
+			}
+
+			if err = validateCORSReplicationPayload(corsData); err != nil {
+				rpt.SetStatus(bucket, fileName, fmt.Errorf("%s (%s)", errorCodes[ErrMalformedXML].Description, err))
+				continue
+			}
+
+			bucketMap[bucket].CorsConfigXML = corsData
+			bucketMap[bucket].CorsConfigUpdatedAt = updatedAt
+			markImported(bucket, fileName)
+			rpt.SetStatus(bucket, fileName, nil)
 		}
 	}
 
@@ -1080,6 +1129,16 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 		if hookNeeded {
 			err = globalSiteReplicationSys.BucketMetaHook(ctx, hook)
 		}
+		if _, ok := fields[bucketCorsConfig]; ok {
+			// CORS carries its own timestamp, so it replicates through the
+			// dedicated event rather than the shared bucket metadata hook. It
+			// is announced even when the shared hook failed: the document is
+			// already committed locally, and a peer that is unreachable for
+			// one config must not withhold CORS from the reachable ones.
+			if corsEvent, live := newBucketCORSReplicationEvent(bucket, *meta); live {
+				err = errors.Join(err, globalSiteReplicationSys.BucketMetaHook(ctx, corsEvent))
+			}
+		}
 		if err != nil {
 			rpt.SetStatus(bucket, "", err)
 			continue

cmd/site-replication.go

@@ -1978,6 +1978,23 @@ func updateLocalBucketCORSMetadata(ctx context.Context, objectAPI ObjectLayer, b
 	return applyBucketCORSMetadata(ctx, objectAPI, bucket, configData, time.Time{}, true)
 }
 
+// localCORSUpdatedAt returns the timestamp a locally originated CORS update
+// must carry: now, advanced strictly past both bucket creation and the current
+// CORS timestamp. A CORS event stamped before bucket creation is discarded as
+// belonging to an older incarnation of the bucket, so a local write must never
+// land at or below that floor.
+func localCORSUpdatedAt(meta BucketMetadata, now time.Time) time.Time {
+	updatedAt := now.UTC()
+	floor := meta.Created
+	if current := meta.CorsConfigUpdatedAt.UTC(); current.After(floor) {
+		floor = current
+	}
+	if !updatedAt.After(floor) {
+		updatedAt = floor.Add(time.Nanosecond)
+	}
+	return updatedAt
+}
+
 func applyBucketCORSMetadata(ctx context.Context, objectAPI ObjectLayer, bucket string, configData []byte, sourceUpdatedAt time.Time, local bool) (time.Time, error) {
 	if bucket == "" || (configData != nil && len(configData) == 0) {
 		return time.Time{}, errInvalidArgument
@@ -2011,14 +2028,7 @@ func applyBucketCORSMetadata(ctx context.Context, objectAPI ObjectLayer, bucket
 	localState := newCORSReplicationState(meta.CorsConfigXML, meta.CorsConfigUpdatedAt)
 	updatedAt := sourceUpdatedAt.UTC()
 	if local {
-		updatedAt = UTCNow()
-		floor := meta.Created
-		if localState.updatedAt.After(floor) {
-			floor = localState.updatedAt
-		}
-		if !updatedAt.After(floor) {
-			updatedAt = floor.Add(time.Nanosecond)
-		}
+		updatedAt = localCORSUpdatedAt(meta, UTCNow())
 	} else {
 		// CreatedAt is the bucket-lineage floor: an event from an older
 		// incarnation of the bucket must not change the current one.