minio/cmd/iam-svcacct-replication-snapshot_test.go

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

package cmd

import (
	"archive/zip"
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"path"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/minio/madmin-go/v3"
	"github.com/minio/minio/internal/auth"
	"github.com/pgsty/silo-pkg/v3/policy"
)

// Site replication sends a service account as the record its site has
// stored, whole. A peer that applies the last item it was sent then has
// the account of the sending site, whatever it received or lost before,
// and whatever the sending server holds in its cache.

const svcAccSnapshotPolicy = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::snapshot/*"]}]}`

// svcAccRecord is what a site has of a service account.
type svcAccRecord struct {
	Parent, Secret, Status, Name, Description, Policy string
	Expiration, UpdatedAt                             time.Time
}

// svcAccPolicyText is a policy document in the one form it is compared in;
// an account without an embedded policy has "".
func svcAccPolicyText(t *testing.T, document []byte) string {
	t.Helper()
	if len(document) == 0 {
		return ""
	}
	p, err := policy.ParseConfig(bytes.NewReader(document))
	mustIAM(t, err)
	text, err := json.Marshal(p)
	mustIAM(t, err)
	return string(text)
}

// svcAccStored is the account as a server that starts now loads it.
func svcAccStored(t *testing.T, f *groupRefreshFixture, key string) svcAccRecord {
	t.Helper()
	u, ok := f.stored(t, key)
	if !ok {
		t.Fatalf("service account %s is not stored", key)
	}
	cr := u.Credentials
	claims, err := getClaimsFromTokenWithSecret(cr.SessionToken, cr.SecretKey)
	mustIAM(t, err)
	rec := svcAccRecord{Parent: cr.ParentUser, Secret: cr.SecretKey, Status: cr.Status, Name: cr.Name, Description: cr.Description, Expiration: cr.Expiration, UpdatedAt: u.UpdatedAt}
	if sp, ok := claims.Lookup(policy.SessionPolicyName); ok && claims.MapClaims[iamPolicyClaimNameSA()] == embeddedPolicyType {
		document, err := base64.StdEncoding.DecodeString(sp)
		mustIAM(t, err)
		rec.Policy = svcAccPolicyText(t, document)
	}
	return rec
}

// svcAccSite is a site that replicates to one peer, which records the IAM
// items it is sent.
type svcAccSite struct {
	*groupRefreshFixture
	mu   sync.Mutex
	sent []iamReplicationItem
}

func newSvcAccSite(t *testing.T) *svcAccSite {
	t.Helper()
	s := &svcAccSite{groupRefreshFixture: newGroupRefreshFixture(t)}
	globalObjLayerMutex.Lock()
	globalObjectAPI = s.obj
	globalObjLayerMutex.Unlock()

	peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method == http.MethodPut && strings.HasSuffix(r.URL.Path, iamRevisionPeerPath) {
			var batch iamRevisionBatch
			if err := json.NewDecoder(r.Body).Decode(&batch); err != nil {
				t.Error(err)
			}
			s.mu.Lock()
			s.sent = append(s.sent, batch.Items...)
			s.mu.Unlock()
		}
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(iamRevisionResponse{iamRevisionStatus: iamRevisionStatus{Version: iamRevisionProtocol, Node: "node", Instance: "boot", Digest: "fixture"}})
	}))
	globalSiteReplicatorCred.Set("snapshot-site-signing-key")
	globalSiteReplicationSys.Lock()
	globalSiteReplicationSys.enabled = true
	globalSiteReplicationSys.state = srState{ServiceAccountAccessKey: s.parent.AccessKey, Peers: s.peers(peer.URL)}
	globalSiteReplicationSys.Unlock()
	t.Cleanup(func() {
		s.leave()
		peer.Close()
	})
	return s
}

func (s *svcAccSite) peers(endpoint string) map[string]madmin.PeerInfo {
	return map[string]madmin.PeerInfo{
		globalDeploymentID(): {Name: "local", DeploymentID: globalDeploymentID()},
		"remote":             {Name: "remote", DeploymentID: "remote", Endpoint: endpoint},
	}
}

// leave ends the replication of this process, for the site that comes next.
func (s *svcAccSite) leave() {
	globalSiteReplicationSys.Lock()
	globalSiteReplicationSys.enabled = false
	globalSiteReplicationSys.state = srState{}
	globalSiteReplicationSys.Unlock()
	globalSiteReplicatorCred.Set("")
}

// items are the creates and edits of account key the peer was sent, in the
// order of sending.
func (s *svcAccSite) items(key string) []iamReplicationItem {
	s.mu.Lock()
	defer s.mu.Unlock()
	var items []iamReplicationItem
	for _, item := range s.sent {
		c := item.SvcAccChange
		if c != nil && ((c.Create != nil && c.Create.AccessKey == key) || (c.Update != nil && c.Update.AccessKey == key)) {
			items = append(items, item)
		}
	}
	return items
}

// create and edit are an administrator's requests to this site.
func (s *svcAccSite) create(t *testing.T, key string) {
	t.Helper()
	rec := iamAdminCall(t, globalActiveCred, http.MethodPut, "/minio/admin/v3/add-service-account",
		madmin.AddServiceAccountReq{TargetUser: s.parent.AccessKey, AccessKey: key, SecretKey: "snapshot-original-secret"}, nil, adminAPIHandlers{}.AddServiceAccount)
	if rec.Code != http.StatusOK {
		t.Fatalf("add-service-account: %d %s", rec.Code, rec.Body.String())
	}
}

func (s *svcAccSite) edit(t *testing.T, key string, req madmin.UpdateServiceAccountReq) {
	t.Helper()
	rec := iamAdminCall(t, globalActiveCred, http.MethodPost, "/minio/admin/v3/update-service-account?accessKey="+key,
		req, map[string]string{"accessKey": key}, adminAPIHandlers{}.UpdateServiceAccount)
	if rec.Code != http.StatusNoContent {
		t.Fatalf("update-service-account: %d %s", rec.Code, rec.Body.String())
	}
}

// svcAccPeer is the site that applies items, in the order given, and what
// it then has of account key. It has the parent user, as a peer does.
func svcAccPeer(t *testing.T, from *svcAccSite, key string, items ...iamReplicationItem) svcAccRecord {
	t.Helper()
	from.leave()
	peer := newGroupRefreshFixture(t)
	for i, item := range items {
		// As the peer reads it off the wire.
		raw, err := json.Marshal(item)
		mustIAM(t, err)
		var received iamReplicationItem
		mustIAM(t, json.Unmarshal(raw, &received))
		if err := applyIAMReplicationItem(peer.ctx, received); err != nil {
			t.Fatalf("the peer refused item %d: %v", i, err)
		}
	}
	return svcAccStored(t, peer, key)
}

func wantSvcAccRecord(t *testing.T, where string, got, want svcAccRecord) {
	t.Helper()
	if got != want {
		t.Errorf("%s:\n got %+v\nwant %+v", where, got, want)
	}
}

// An edit of a service account reaches a peer whatever became of the edits
// before it: one that arrives late, or never, is not lost.
func TestSiteReplicationServiceAccountEdits(t *testing.T) {
	const key = "snapshot-edits"
	for _, tc := range []struct {
		name  string
		edits []madmin.UpdateServiceAccountReq
		took  func(svcAccRecord) bool // control: what the site has after them
	}{
		{"policy-secret-description", []madmin.UpdateServiceAccountReq{
			{NewPolicy: json.RawMessage(svcAccSnapshotPolicy)},
			{NewSecretKey: "snapshot-rotated-secret"},
			{NewDescription: "described"},
		}, func(r svcAccRecord) bool {
			return r.Policy != "" && r.Secret == "snapshot-rotated-secret" && r.Description == "described"
		}},
		{"secret-disable", []madmin.UpdateServiceAccountReq{
			{NewPolicy: json.RawMessage(svcAccSnapshotPolicy), NewSecretKey: "snapshot-rotated-secret"},
			{NewStatus: string(madmin.AccountDisabled)},
		}, func(r svcAccRecord) bool {
			return r.Policy != "" && r.Secret == "snapshot-rotated-secret" && r.Status == auth.AccountOff
		}},
		// The last edit removes the embedded policy: the peer gets an
		// account without one, and the description set before.
		{"policy-removed", []madmin.UpdateServiceAccountReq{
			{NewPolicy: json.RawMessage(svcAccSnapshotPolicy)},
			{NewDescription: "described"},
			{NewPolicy: json.RawMessage(`{"Statement":[]}`)},
		}, func(r svcAccRecord) bool { return r.Policy == "" && r.Description == "described" }},
		// Policy documents an edit ignores: one with a version and no
		// statement, one with statements and no version. The site keeps the
		// policy it has, and so does the peer.
		{"policy-ignored", []madmin.UpdateServiceAccountReq{
			{NewPolicy: json.RawMessage(svcAccSnapshotPolicy)},
			{NewPolicy: json.RawMessage(`{"Version":"2012-10-17","Statement":[]}`), NewName: "named"},
			{NewPolicy: json.RawMessage(`{"Statement":[{"Effect":"Allow","Action":["s3:PutObject"],"Resource":["arn:aws:s3:::ignored/*"]}]}`), NewDescription: "described"},
		}, func(r svcAccRecord) bool {
			return r.Policy == svcAccPolicyText(t, []byte(svcAccSnapshotPolicy)) && r.Name == "named" && r.Description == "described"
		}},
	} {
		for _, delivery := range []string{"late", "lost"} {
			t.Run(tc.name+"/"+delivery, func(t *testing.T) {
				site := newSvcAccSite(t)
				site.create(t, key)
				for _, edit := range tc.edits {
					site.edit(t, key, edit)
				}
				sent := site.items(key)
				if len(sent) != 1+len(tc.edits) {
					t.Fatalf("the site sent %d items for the account, want one for the create and each edit", len(sent))
				}
				want := svcAccStored(t, site.groupRefreshFixture, key)
				if !tc.took(want) {
					t.Fatalf("control: the site did not apply the edits: %+v", want)
				}
				// The peer has the account and gets the last edit. The edits
				// before it were lost, or reach it after the last.
				last := len(sent) - 1
				deliver := []iamReplicationItem{sent[0], sent[last]}
				if delivery == "late" {
					deliver = append(deliver, sent[1:last]...)
				}
				wantSvcAccRecord(t, "peer", svcAccPeer(t, site, key, deliver...), want)
			})
		}
	}

	// Two requests edit the account through this site at once, and the
	// first comes to replicate when the second is stored.
	t.Run("concurrent", func(t *testing.T) {
		site := newSvcAccSite(t)
		site.create(t, key)
		sp, err := policy.ParseConfig(strings.NewReader(svcAccSnapshotPolicy))
		mustIAM(t, err)
		first, err := site.sys.UpdateServiceAccount(site.ctx, key, updateServiceAccountOpts{sessionPolicy: sp})
		mustIAM(t, err)
		second, err := site.sys.UpdateServiceAccount(site.ctx, key, updateServiceAccountOpts{description: "described"})
		mustIAM(t, err)
		for _, at := range []time.Time{first, second} {
			mustIAM(t, globalSiteReplicationSys.IAMChangeHook(site.ctx, madmin.SRIAMItem{
				Type:         madmin.SRIAMItemSvcAcc,
				SvcAccChange: &madmin.SRSvcAccChange{Update: &madmin.SRSvcAccUpdate{AccessKey: key}},
				UpdatedAt:    at,
			}))
		}
		want := svcAccStored(t, site.groupRefreshFixture, key)
		if want.Policy == "" || want.Description != "described" {
			t.Fatalf("control: the site did not apply the edits: %+v", want)
		}
		wantSvcAccRecord(t, "peer", svcAccPeer(t, site, key, site.items(key)...), want)
	})
}

// Another server of the site restricted, rotated and disabled the account,
// and this server's cache has not heard of it. What this server sends, and
// exports, is the account that is stored.
func TestSiteReplicationServiceAccountStaleCache(t *testing.T) {
	const key = "snapshot-stale"
	// prepare returns the site, the stored account and what it sent for the
	// create.
	prepare := func(t *testing.T) (*svcAccSite, svcAccRecord, iamReplicationItem) {
		t.Helper()
		site := newSvcAccSite(t)
		site.create(t, key)
		created := site.items(key)
		if len(created) != 1 {
			t.Fatalf("the site sent %d items for the create", len(created))
		}
		sp, err := policy.ParseConfig(strings.NewReader(svcAccSnapshotPolicy))
		mustIAM(t, err)
		_, err = site.sibling(t).UpdateServiceAccount(site.ctx, key, updateServiceAccountOpts{
			sessionPolicy: sp, secretKey: "snapshot-rotated-secret", status: auth.AccountOff, description: "restricted",
		})
		mustIAM(t, err)
		if cached, _ := site.sys.store.GetUser(key); cached.Credentials.SecretKey != "snapshot-original-secret" || cached.Credentials.Status != auth.AccountOn {
			t.Fatal("control: this server's cache must not have heard of the edit")
		}
		want := svcAccStored(t, site.groupRefreshFixture, key)
		if want.Policy == "" || want.Secret != "snapshot-rotated-secret" || want.Status != auth.AccountOff {
			t.Fatalf("control: the edit is not stored: %+v", want)
		}
		return site, want, created[0]
	}

	t.Run("heal", func(t *testing.T) {
		site, want, created := prepare(t)
		local := madmin.UserInfo{Status: madmin.AccountStatus(want.Status), UpdatedAt: want.UpdatedAt}
		remote := madmin.UserInfo{Status: madmin.AccountStatus(auth.AccountOn), UpdatedAt: created.UpdatedAt}
		info := srStatusInfo{Sites: globalSiteReplicationSys.state.Peers, UserStats: map[string]map[string]srUserStatsSummary{key: {
			globalDeploymentID(): {userInfo: srUserInfo{UserInfo: local}},
			"remote":             {SRUserStatsSummary: madmin.SRUserStatsSummary{UserInfoMismatch: true}, userInfo: srUserInfo{UserInfo: remote}},
		}}}
		mustIAM(t, globalSiteReplicationSys.healUsers(site.ctx, site.obj, key, info))
		sent := site.items(key)
		if len(sent) != 2 {
			t.Fatalf("heal sent %d items for the account, want 1", len(sent)-1)
		}
		wantSvcAccRecord(t, "healed peer", svcAccPeer(t, site, key, sent...), want)
	})

	t.Run("sync", func(t *testing.T) {
		site, want, _ := prepare(t)
		mustIAM(t, globalSiteReplicationSys.syncToAllPeers(site.ctx, madmin.SRAddOptions{}))
		sent := site.items(key)
		if len(sent) != 2 {
			t.Fatalf("sync sent %d items for the account, want 1", len(sent)-1)
		}
		wantSvcAccRecord(t, "new peer", svcAccPeer(t, site, key, sent[1]), want)
	})

	t.Run("export", func(t *testing.T) {
		_, want, _ := prepare(t)
		rec := iamAdminCall(t, globalActiveCred, http.MethodGet, "/minio/admin/v3/export-iam", nil, nil, adminAPIHandlers{}.ExportIAM)
		if rec.Code != http.StatusOK {
			t.Fatalf("export-iam: %d %s", rec.Code, rec.Body.String())
		}
		archive, err := zip.NewReader(bytes.NewReader(rec.Body.Bytes()), int64(rec.Body.Len()))
		mustIAM(t, err)
		file, err := archive.Open(path.Join(iamAssetsDir, allSvcAcctsFile))
		mustIAM(t, err)
		raw, err := io.ReadAll(file)
		mustIAM(t, err)
		var accounts map[string]madmin.SRSvcAccCreate
		mustIAM(t, json.Unmarshal(raw, &accounts))
		a, ok := accounts[key]
		if !ok || a.Expiration == nil {
			t.Fatalf("the export has no account %s with an expiration (present=%v)", key, ok)
		}
		got := svcAccRecord{Parent: a.Parent, Secret: a.SecretKey, Status: a.Status, Name: a.Name, Description: a.Description, Policy: svcAccPolicyText(t, a.SessionPolicy), Expiration: *a.Expiration}
		want.UpdatedAt = time.Time{} // an export has no revisions
		wantSvcAccRecord(t, "export", got, want)
		if a.Claims[iamPolicyClaimNameSA()] != embeddedPolicyType {
			t.Errorf("exported claims say %v, want %s", a.Claims[iamPolicyClaimNameSA()], embeddedPolicyType)
		}
	})
}

// The parent was deleted and created again on both sites. The token of a
// service account created afterwards names that revocation, and a peer
// refuses a credential of the parent that does not: the create must carry
// the claims of the token.
func TestSiteReplicationServiceAccountOfRecreatedParent(t *testing.T) {
	const key = "snapshot-reparent"
	site := newSvcAccSite(t)
	mustIAM(t, site.sys.DeleteUser(site.ctx, site.parent.AccessKey, false))
	iamAdminUser(site.ctx, t, site.sys, site.parent.AccessKey, "refresh-parent")
	revoked, err := loadIAMRevision(site.ctx, site.sys.store, getUserIdentityPath(site.parent.AccessKey, regUser))
	mustIAM(t, err)
	if revoked.RevokedBefore.IsZero() {
		t.Fatal("control: the parent must have a revocation boundary")
	}
	site.create(t, key)
	sent := site.items(key)
	if len(sent) != 1 {
		t.Fatalf("the site sent %d items for the create", len(sent))
	}
	want := svcAccStored(t, site.groupRefreshFixture, key)

	// The peer's parent is newer than the deletion, which leaves it live
	// with the boundary: a parent that was deleted and created again.
	site.leave()
	peer := newGroupRefreshFixture(t)
	mustIAM(t, peer.sys.DeleteUser(withIAMReplicationTime(peer.ctx, revoked.RevokedBefore), peer.parent.AccessKey, false))
	if r, err := loadIAMRevision(peer.ctx, peer.sys.store, getUserIdentityPath(peer.parent.AccessKey, regUser)); err != nil || r.Deleted || !r.RevokedBefore.Equal(revoked.RevokedBefore) {
		t.Fatalf("control: the peer's parent must be live with the boundary (err=%v deleted=%v)", err, r.Deleted)
	}
	mustIAM(t, applyIAMReplicationItem(peer.ctx, sent[0]))
	if _, ok := peer.stored(t, key); !ok {
		t.Fatal("the peer dropped the new service account")
	}
	wantSvcAccRecord(t, "peer", svcAccStored(t, peer, key), want)
}

// An account that is deleted is not sent, and that is not an error. An
// account whose stored token cannot be read is not taken for deleted: the
// site signing key is not available here, and no token is read without it.
func TestSiteReplicationServiceAccountUnreadable(t *testing.T) {
	const key = "snapshot-unreadable"
	site := newSvcAccSite(t)
	site.create(t, key)
	item := madmin.SRIAMItem{Type: madmin.SRIAMItemSvcAcc, SvcAccChange: &madmin.SRSvcAccChange{Update: &madmin.SRSvcAccUpdate{AccessKey: key}}}

	globalSiteReplicatorCred.Set("")
	if _, err := globalSiteReplicationSys.replicationItem(site.ctx, item); err == nil || errors.Is(err, errIAMStaleUpdate) {
		t.Errorf("an account whose token cannot be read must be an error, got %v", err)
	}
	if err := globalSiteReplicationSys.IAMChangeHook(site.ctx, item); err == nil {
		t.Error("the hook reported an account as sent that it could not read")
	}
	globalSiteReplicatorCred.Set("snapshot-site-signing-key")

	mustIAM(t, site.sys.DeleteServiceAccount(site.ctx, key, false))
	if _, err := globalSiteReplicationSys.replicationItem(site.ctx, item); !errors.Is(err, errIAMStaleUpdate) {
		t.Errorf("a deleted account must not be sent, got %v", err)
	}
	if sent := site.items(key); len(sent) != 1 {
		t.Errorf("the site sent %d items for the account, want only the create", len(sent))
	}
}

// svcAccPausedStore holds every load of a user until proceed is closed.
type svcAccPausedStore struct {
	IAMStorageAPI
	once             sync.Once
	loading, proceed chan struct{}
}

func (s *svcAccPausedStore) loadUser(ctx context.Context, user string, userType IAMUserType, m map[string]UserIdentity) error {
	s.once.Do(func() { close(s.loading) })
	<-s.proceed
	return s.IAMStorageAPI.loadUser(ctx, user, userType, m)
}

// The hook reads the account from storage without the lock of the
// replication configuration. Loading an identity takes that lock for
// reading; if the hook held it too, a change of the configuration that
// arrives in between would wait for the hook, and the hook for it.
func TestSiteReplicationHookReadsStorageUnlocked(t *testing.T) {
	const key = "snapshot-unlocked"
	site := newSvcAccSite(t)
	site.create(t, key)
	paused := &svcAccPausedStore{IAMStorageAPI: site.sys.store.IAMStorageAPI, loading: make(chan struct{}), proceed: make(chan struct{})}
	site.sys.store.IAMStorageAPI = paused

	sent := make(chan error, 1)
	go func() {
		sent <- globalSiteReplicationSys.IAMChangeHook(site.ctx, madmin.SRIAMItem{
			Type:         madmin.SRIAMItemSvcAcc,
			SvcAccChange: &madmin.SRSvcAccChange{Update: &madmin.SRSvcAccUpdate{AccessKey: key}},
		})
	}()
	select {
	case <-paused.loading:
	case err := <-sent:
		t.Fatalf("the hook did not read the account from storage (err=%v)", err)
	}
	// A writer that had to wait here would never get the lock, nor the
	// hook its next read lock behind that writer: try, do not wait.
	if globalSiteReplicationSys.TryLock() {
		globalSiteReplicationSys.Unlock() //nolint:staticcheck // only the lock is of interest
	} else {
		t.Error("the hook holds the lock of the replication configuration while it reads storage")
	}
	close(paused.proceed)
	select {
	case err := <-sent:
		mustIAM(t, err)
	case <-time.After(10 * time.Second):
		t.Fatal("the hook did not return")
	}
	if got := site.items(key); len(got) != 2 {
		t.Errorf("the site sent %d items for the account, want the create and the edit", len(got))
	}
}