minio/cmd/iam-replication-protocol_test.go

// Copyright (c) 2026 PGSTY
// SPDX-License-Identifier: AGPL-3.0-or-later

package cmd

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/minio/madmin-go/v3"
)

func TestIAMRevisionProtocolDoesNotFallBackToLegacy(t *testing.T) {
	var requests atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		requests.Add(1)
		if r.URL.Path != "/minio/admin/v3/site-replication/peer/iam-revisions" {
			t.Errorf("unsafe fallback path: %s", r.URL.Path)
		}
		w.WriteHeader(http.StatusNotFound)
		_, _ = w.Write([]byte(`{"Code":"NotImplemented","Message":"old server"}`))
	}))
	defer server.Close()
	client, err := madmin.New(strings.TrimPrefix(server.URL, "http://"), "test-access", "valid-test-secret", false)
	mustIAM(t, err)
	_, err = executeIAMRevisionRequest(context.Background(), client, http.MethodPut, &iamRevisionBatch{Version: iamRevisionProtocol, Items: []iamReplicationItem{{SRIAMItem: madmin.SRIAMItem{Type: iamUserBoundaryType}, UserRevocation: &iamUserBoundary{User: "recreated", Before: UTCNow()}}}})
	if err == nil || requests.Load() != 1 {
		t.Fatalf("old peer must reject without fallback, err=%v requests=%d", err, requests.Load())
	}
}

type iamNoHealingScanStore struct{ IAMStorageAPI }

func (s *iamNoHealingScanStore) listIAMConfigPaths(context.Context) ([]string, error) {
	panic("healing must use the loaded revision index")
}

func TestIAMRevisionHealingAcknowledgements(t *testing.T) {
	for _, balanced := range []bool{false, true} {
		t.Run(fmt.Sprintf("load_balanced_%t", balanced), func(t *testing.T) { testIAMRevisionHealingAcknowledgements(t, balanced) })
	}
}

func testIAMRevisionHealingAcknowledgements(t *testing.T, balanced bool) {
	ctx, sys, _ := prepareIAMRevisionFixture(t)
	_, err := sys.CreateUser(ctx, "ack-sync", madmin.AddOrUpdateUserReq{SecretKey: "valid-sync-password", Status: madmin.AccountEnabled})
	mustIAM(t, err)
	for i := range maxIAMRevisionBatch*2 + 1 {
		at := UTCNow().Add(time.Duration(i) * time.Nanosecond)
		mustIAM(t, sys.store.saveIAMConfig(ctx, &UserIdentity{Version: 1, Deleted: true, UpdatedAt: at, RevokedBefore: at}, getUserIdentityPath(fmt.Sprintf("ack-%04d", i), regUser)))
	}
	sys.store.IAMStorageAPI = &iamNoHealingScanStore{IAMStorageAPI: sys.store.IAMStorageAPI}
	var mu sync.Mutex
	var puts, gets int
	var applied int
	instance := "boot-1"
	failSecondBatch := true
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path == "/minio/health/live" {
			w.WriteHeader(http.StatusOK)
			return
		}
		mu.Lock()
		defer mu.Unlock()
		if r.URL.Path != "/minio/admin/v3/site-replication/peer/iam-revisions" {
			t.Errorf("unexpected request: %s", r.URL.Path)
			w.WriteHeader(404)
			return
		}
		var failures []string
		if r.Method == http.MethodGet {
			gets++
		} else {
			puts++
			var batch iamRevisionBatch
			if err := json.NewDecoder(r.Body).Decode(&batch); err != nil {
				t.Error(err)
				w.WriteHeader(400)
				return
			}
			if len(batch.Items) > maxIAMRevisionBatch {
				t.Error("batch exceeds limit")
			}
			if failSecondBatch && puts == 2 {
				failures = []string{"injected item error"}
			} else {
				applied += len(batch.Items)
			}
		}
		node := "node-1"
		if balanced {
			node = fmt.Sprintf("node-%d", (gets+puts)%2+1)
		}
		_ = json.NewEncoder(w).Encode(iamRevisionResponse{iamRevisionStatus: iamRevisionStatus{Version: iamRevisionProtocol, Node: node, Instance: node + instance, Digest: fmt.Sprintf("%d", applied)}, Errors: failures})
	}))
	defer server.Close()
	c := &SiteReplicationSys{enabled: true, state: srState{ServiceAccountAccessKey: "ack-sync", Peers: map[string]madmin.PeerInfo{globalDeploymentID(): {DeploymentID: globalDeploymentID(), Name: "local"}, "remote": {DeploymentID: "remote", Name: "remote", Endpoint: server.URL}}}}
	if err := c.healIAMDeletions(ctx); err == nil {
		t.Fatal("item failure was hidden")
	}
	mu.Lock()
	if puts != 3 || applied != maxIAMRevisionBatch+1 {
		t.Errorf("failed middle batch blocked later revocations: puts=%d applied=%d", puts, applied)
	}
	failSecondBatch = false
	applied++ // Unrelated remote mutation changes its digest.
	mu.Unlock()
	at := UTCNow()
	mustIAM(t, sys.store.saveIAMConfig(ctx, &UserIdentity{Version: 1, Deleted: true, UpdatedAt: at, RevokedBefore: at}, getUserIdentityPath("ack-new-local", regUser)))
	mustIAM(t, c.healIAMDeletions(ctx))
	mu.Lock()
	if puts != 5 {
		t.Errorf("did not resume at unacknowledged batch: puts=%d", puts)
	}
	mu.Unlock()
	mustIAM(t, c.healIAMDeletions(ctx))
	mu.Lock()
	if puts != 5 || gets != 3 {
		t.Errorf("converged records were replayed: puts=%d gets=%d", puts, gets)
	}
	instance = "boot-2"
	mu.Unlock()
	mustIAM(t, c.healIAMDeletions(ctx))
	mu.Lock()
	defer mu.Unlock()
	if puts != 8 {
		t.Fatalf("peer restart reused an old acknowledgement: puts=%d", puts)
	}
}

func TestIAMRevisionIndexRebuildsFromStorage(t *testing.T) {
	ctx, sys, obj := prepareIAMRevisionFixture(t)
	const user = "index-parent"
	req := madmin.AddOrUpdateUserReq{SecretKey: "valid-parent-password", Status: madmin.AccountEnabled}
	_, err := sys.CreateUser(ctx, user, req)
	mustIAM(t, err)
	mustIAM(t, sys.DeleteUser(ctx, user, false))
	before := sys.store.revisionIndex().snapshot()
	store := &IAMStoreSys{IAMStorageAPI: newIAMObjectStore(obj, MinIOUsersSysType)}
	mustIAM(t, store.LoadIAMCache(ctx, true))
	if iamRevisionDigest(before) != iamRevisionDigest(store.revisionIndex().snapshot()) {
		t.Fatal("ordinary IAM loading did not restore the deletion index")
	}
	_, err = store.AddUser(ctx, user, req)
	mustIAM(t, err)
	r := store.revisionIndex().get(getUserIdentityPath(user, regUser))
	if r.Deleted || r.RevokedBefore.IsZero() {
		t.Fatal("recreation discarded the retained boundary")
	}
	if r.Credentials.SecretKey != "" || r.Credentials.SessionToken != "" {
		t.Fatal("index retained credentials")
	}
}