Merge remote-tracking branch 'origin/main' into codex/pr257-delete-safety

Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/config-migrate.go

@@ -91,7 +91,7 @@ func readConfigWithoutMigrate(ctx context.Context, objAPI ObjectLayer) (config.C
 			return nil, err
 		}
 
-		data, err = decryptData(data, configFile)
+		data, err = decryptData(ctx, data, configFile)
 		if err != nil {
 			return nil, err
 		}

cmd/config.go

@@ -68,7 +68,7 @@ func listServerConfigHistory(ctx context.Context, objAPI ObjectLayer, withData b
 					continue
 				}
 
-				data, err = decryptData(data, obj.Name)
+				data, err = decryptData(ctx, data, obj.Name)
 				if err != nil {
 					// ignore history file that cannot be loaded.
 					continue
@@ -110,7 +110,7 @@ func readServerConfigHistory(ctx context.Context, objAPI ObjectLayer, uuidKV str
 		return nil, err
 	}
 
-	return decryptData(data, historyFile)
+	return decryptData(ctx, data, historyFile)
 }
 
 func saveServerConfigHistory(ctx context.Context, objAPI ObjectLayer, kv []byte) error {
@@ -162,7 +162,7 @@ func readServerConfig(ctx context.Context, objAPI ObjectLayer, data []byte) (con
 			return nil, err
 		}
 
-		data, err = decryptData(data, configFile)
+		data, err = decryptData(ctx, data, configFile)
 		if err != nil {
 			lookupConfigs(srvCfg, objAPI)
 			return nil, err

cmd/iam-etcd-store.go

@@ -123,8 +123,8 @@ func (ies *IAMEtcdStore) saveIAMConfig(ctx context.Context, item any, itemPath s
 	return nil
 }
 
-func (ies *IAMEtcdStore) decodeIAMConfig(item any, data []byte, path string) error {
-	data, err := decryptData(data, path)
+func (ies *IAMEtcdStore) decodeIAMConfig(ctx context.Context, item any, data []byte, path string) error {
+	data, err := decryptData(ctx, data, path)
 	if err != nil {
 		return err
 	}
@@ -145,7 +145,7 @@ func (ies *IAMEtcdStore) loadIAMConfigBytes(ctx context.Context, path string) ([
 	if err != nil {
 		return nil, err
 	}
-	data, err = decryptData(data, path)
+	data, err = decryptData(ctx, data, path)
 	if err == nil {
 		ies.index.observe(path, data)
 	}
@@ -187,7 +187,7 @@ func (ies *IAMEtcdStore) loadPolicyDoc(ctx context.Context, policy string, m map
 }
 
 func (ies *IAMEtcdStore) getPolicyDocKV(ctx context.Context, kvs *mvccpb.KeyValue, m map[string]PolicyDoc) error {
-	data, err := decryptData(kvs.Value, string(kvs.Key))
+	data, err := decryptData(ctx, kvs.Value, string(kvs.Key))
 	if err != nil {
 		if err == errConfigNotFound {
 			return errNoSuchPolicy
@@ -211,27 +211,31 @@ func (ies *IAMEtcdStore) getPolicyDocKV(ctx context.Context, kvs *mvccpb.KeyValu
 }
 
 func (ies *IAMEtcdStore) loadPolicyDocs(ctx context.Context, m map[string]PolicyDoc) error {
-	ctx, cancel := context.WithTimeout(ctx, defaultContextTimeout)
+	// Bound the etcd read without shortening the caller's decryption budget.
+	cctx, cancel := context.WithTimeout(ctx, defaultContextTimeout)
 	defer cancel()
 	//  Retrieve all keys and values to avoid too many calls to etcd in case of
 	//  a large number of policies
-	r, err := ies.client.Get(ctx, iamConfigPoliciesPrefix, etcd.WithPrefix())
+	r, err := ies.client.Get(cctx, iamConfigPoliciesPrefix, etcd.WithPrefix())
 	if err != nil {
 		return err
 	}
 
 	// Parse all values to construct the policies data model.
 	for _, kvs := range r.Kvs {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
 		if err = ies.getPolicyDocKV(ctx, kvs, m); err != nil && !errors.Is(err, errNoSuchPolicy) {
 			return err
 		}
 	}
-	return nil
+	return ctx.Err()
 }
 
 func (ies *IAMEtcdStore) getUserKV(ctx context.Context, userkv *mvccpb.KeyValue, userType IAMUserType, m map[string]UserIdentity, basePrefix string) error {
 	var u UserIdentity
-	err := ies.decodeIAMConfig(&u, userkv.Value, string(userkv.Key))
+	err := ies.decodeIAMConfig(ctx, &u, userkv.Value, string(userkv.Key))
 	if err != nil {
 		if err == errConfigNotFound {
 			return errNoSuchUser
@@ -392,9 +396,9 @@ func (ies *IAMEtcdStore) loadMappedPolicy(ctx context.Context, name string, user
 	return nil
 }
 
-func (ies *IAMEtcdStore) getMappedPolicy(kv *mvccpb.KeyValue, m *xsync.MapOf[string, MappedPolicy], basePrefix string) error {
+func (ies *IAMEtcdStore) getMappedPolicy(ctx context.Context, kv *mvccpb.KeyValue, m *xsync.MapOf[string, MappedPolicy], basePrefix string) error {
 	var p MappedPolicy
-	err := ies.decodeIAMConfig(&p, kv.Value, string(kv.Key))
+	err := ies.decodeIAMConfig(ctx, &p, kv.Value, string(kv.Key))
 	if err != nil {
 		if err == errConfigNotFound {
 			return errNoSuchPolicy
@@ -434,7 +438,7 @@ func (ies *IAMEtcdStore) loadMappedPolicies(ctx context.Context, userType IAMUse
 
 	// Parse all policies mapping to create the proper data model
 	for _, kv := range r.Kvs {
-		if err = ies.getMappedPolicy(kv, m, basePrefix); err != nil && !errors.Is(err, errNoSuchPolicy) {
+		if err = ies.getMappedPolicy(ctx, kv, m, basePrefix); err != nil && !errors.Is(err, errNoSuchPolicy) {
 			return err
 		}
 	}

cmd/iam-object-store.go

@@ -108,7 +108,10 @@ func (iamOS *IAMObjectStore) saveIAMConfig(ctx context.Context, item any, objPat
 	return nil
 }
 
-func decryptData(data []byte, objPath string) ([]byte, error) {
+func decryptData(ctx context.Context, data []byte, objPath string) ([]byte, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	if utf8.Valid(data) {
 		return data, nil
 	}
@@ -118,13 +121,17 @@ func decryptData(data []byte, objPath string) ([]byte, error) {
 		return pdata, nil
 	}
 	if GlobalKMS != nil {
-		pdata, err = config.DecryptBytes(GlobalKMS, data, kms.Context{
+		pdata, err = config.DecryptBytes(ctx, GlobalKMS, data, kms.Context{
 			minioMetaBucket: path.Join(minioMetaBucket, objPath),
 		})
 		if err == nil {
 			return pdata, nil
 		}
-		pdata, err = config.DecryptBytes(GlobalKMS, data, kms.Context{
+		// Cancellation must not trigger the legacy associated-data fallback.
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		pdata, err = config.DecryptBytes(ctx, GlobalKMS, data, kms.Context{
 			minioMetaBucket: objPath,
 		})
 		if err == nil {
@@ -164,7 +171,7 @@ func (iamOS *IAMObjectStore) loadIAMConfigBytesWithMetadata(ctx context.Context,
 		}
 		return nil, meta, err
 	}
-	data, err = decryptData(data, objPath)
+	data, err = decryptData(ctx, data, objPath)
 	if err != nil {
 		return nil, meta, err
 	}
@@ -276,12 +283,15 @@ func (iamOS *IAMObjectStore) loadPolicyDocs(ctx context.Context, m map[string]Po
 			return item.Err
 		}
 
+		if err := ctx.Err(); err != nil {
+			return err
+		}
 		policyName := path.Dir(item.Item)
 		if err := iamOS.loadPolicyDoc(ctx, policyName, m); err != nil && !errors.Is(err, errNoSuchPolicy) {
 			return err
 		}
 	}
-	return nil
+	return ctx.Err()
 }
 
 func (iamOS *IAMObjectStore) loadSecretKey(ctx context.Context, user string, userType IAMUserType) (string, error) {

cmd/iam-policy-budget_test.go

@@ -0,0 +1,181 @@
+// Copyright (c) 2026 Feng Ruohang
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/config"
+	"github.com/minio/minio/internal/kms"
+	"github.com/puzpuzpuz/xsync/v3"
+	etcd "go.etcd.io/etcd/client/v3"
+)
+
+type iamPolicyBudgetGetObserver struct {
+	etcd.KV
+	elapsed time.Duration
+	budget  time.Duration
+	block   bool
+}
+
+func (k *iamPolicyBudgetGetObserver) Get(ctx context.Context, key string, opts ...etcd.OpOption) (*etcd.GetResponse, error) {
+	start := time.Now()
+	if d, ok := ctx.Deadline(); ok {
+		k.budget = time.Until(d)
+	}
+	defer func() { k.elapsed = time.Since(start) }()
+	if k.block {
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+	return k.KV.Get(ctx, key, opts...)
+}
+
+func iamPolicyBudgetCall(ctx context.Context, store *IAMStoreSys, operation string) (int, error) {
+	switch operation {
+	case "policies":
+		m, err := store.ListPolicies(ctx, "")
+		return len(m), err
+	case "docs":
+		m, err := store.ListPolicyDocs(ctx, "")
+		return len(m), err
+	default:
+		m := xsync.NewMapOf[string, MappedPolicy]()
+		err := store.IAMStorageAPI.(*IAMEtcdStore).loadMappedPolicies(ctx, regUser, false, m)
+		return m.Size(), err
+	}
+}
+
+func prepareIAMPolicyBudgetFixture(t *testing.T, operation string) (*IAMStoreSys, *iamPolicyBudgetGetObserver, *atomic.Int32) {
+	t.Helper()
+	ctx, sys, _ := prepareIAMRevisionFixture(t, "etcd")
+	backend := sys.store.IAMStorageAPI.(*IAMEtcdStore)
+	builtin, err := kms.NewBuiltin("cancellation-key", bytes.Repeat([]byte{43}, 32))
+	mustIAM(t, err)
+	var doc any
+	path := getPolicyDocPath("budget-slow")
+	if operation == "mappings" {
+		path = getMappedPolicyPath("budget-slow", regUser, false)
+		doc = newMappedPolicy("readonly")
+	} else {
+		p, err := sys.store.GetPolicy("readonly")
+		mustIAM(t, err)
+		doc = newPolicyDoc(p)
+	}
+	plain, err := json.Marshal(doc)
+	mustIAM(t, err)
+	encrypted, err := config.EncryptBytes(builtin, plain, kms.Context{minioMetaBucket: minioMetaBucket + "/" + path})
+	mustIAM(t, err)
+	_, err = backend.client.Put(ctx, path, string(encrypted))
+	mustIAM(t, err)
+	calls := &atomic.Int32{}
+	priorKMS := GlobalKMS
+	GlobalKMS = connectCancellationKES(t, cancellationDecryptHandler(t, builtin, func(r *http.Request) {
+		calls.Add(1)
+		select {
+		case <-time.After(450 * time.Millisecond):
+		case <-r.Context().Done():
+		}
+	}))
+	t.Cleanup(func() { GlobalKMS = priorKMS })
+	priorTimeout := defaultContextTimeout
+	defaultContextTimeout = 200 * time.Millisecond
+	t.Cleanup(func() { defaultContextTimeout = priorTimeout })
+	observer := &iamPolicyBudgetGetObserver{KV: backend.client.KV}
+	backend.client.KV = observer
+	t.Cleanup(func() { backend.client.KV = observer.KV })
+	return sys.store, observer, calls
+}
+
+// A fast etcd Get followed by slow KES decryption must retain the caller's
+// budget, even when decryption lasts longer than the etcd read timeout.
+func TestIAMPolicyEtcdDecodeBudget(t *testing.T) {
+	for _, operation := range []string{"policies", "docs", "mappings"} {
+		t.Run(operation, func(t *testing.T) {
+			store, observer, calls := prepareIAMPolicyBudgetFixture(t, operation)
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			start := time.Now()
+			n, err := iamPolicyBudgetCall(ctx, store, operation)
+			t.Logf("operation=%s total_ms=%.3f get_ms=%.3f get_budget_ms=%.3f decrypts=%d count=%d error=%v caller_error=%v", operation, float64(time.Since(start).Microseconds())/1000, float64(observer.elapsed.Microseconds())/1000, float64(observer.budget.Microseconds())/1000, calls.Load(), n, err, ctx.Err())
+			if err != nil {
+				t.Fatalf("valid caller budget was shortened after etcd Get completed: %v", err)
+			}
+			if observer.elapsed >= 200*time.Millisecond || observer.budget <= 0 || observer.budget > 210*time.Millisecond {
+				t.Fatalf("etcd read budget invalid: %+v", observer)
+			}
+			if n == 0 || calls.Load() != 1 || ctx.Err() != nil {
+				t.Fatal("successful result was incomplete")
+			}
+		})
+	}
+}
+
+// The correction must retain bounded etcd RPCs and caller cancellation. A
+// failed list cannot publish its private map or advance the cache timestamp.
+func TestIAMPolicyEtcdBudgetGuards(t *testing.T) {
+	for _, operation := range []string{"policies", "docs", "mappings"} {
+		for _, mode := range []string{"read_timeout", "caller_deadline", "caller_cancel"} {
+			t.Run(fmt.Sprintf("%s/%s", operation, mode), func(t *testing.T) {
+				store, observer, calls := prepareIAMPolicyBudgetFixture(t, operation)
+				cache := store.rlock()
+				before, count := cache.updatedAt, len(cache.iamPolicyDocsMap)
+				store.runlock()
+				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+				defer cancel()
+				want := context.DeadlineExceeded
+				if mode == "read_timeout" {
+					observer.block = true
+				}
+				if mode == "caller_deadline" {
+					cancel()
+					ctx, cancel = context.WithTimeout(context.Background(), 80*time.Millisecond)
+					defer cancel()
+				}
+				var timer *time.Timer
+				if mode == "caller_cancel" {
+					timer = time.AfterFunc(80*time.Millisecond, cancel)
+					defer timer.Stop()
+					want = context.Canceled
+				}
+				start := time.Now()
+				_, err := iamPolicyBudgetCall(ctx, store, operation)
+				t.Logf("operation=%s mode=%s elapsed_ms=%.3f decrypts=%d error=%v", operation, mode, float64(time.Since(start).Microseconds())/1000, calls.Load(), err)
+				if !errors.Is(err, want) {
+					t.Fatalf("expected %v, got %v", want, err)
+				}
+				if mode == "read_timeout" && (calls.Load() != 0 || ctx.Err() != nil) {
+					t.Fatal("read timeout did not remain local to the etcd RPC")
+				}
+				cache = store.rlock()
+				unchanged := cache.updatedAt.Equal(before) && len(cache.iamPolicyDocsMap) == count
+				store.runlock()
+				if !unchanged {
+					t.Fatal("failed list published a cache replacement")
+				}
+			})
+		}
+	}
+}

cmd/iam-policy-refresh_test.go

@@ -0,0 +1,182 @@
+// Copyright (c) 2026 Feng Ruohang
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/auth"
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+type iamPolicyRefreshContext struct{}
+
+// Keep IAMStorageAPI as *IAMObjectStore so LoadIAMCache exercises the actual
+// 32-worker object loader. Capture bytes and release the physical read lock
+// before pausing; concurrent writes can commit against real object storage.
+type iamPolicyRefreshReadGate struct {
+	ObjectLayer
+	policyPath, userPath                                     string
+	listEntered, listRelease, refreshEntered, refreshRelease chan struct{}
+	listOnce, refreshOnce                                    sync.Once
+}
+
+func (g *iamPolicyRefreshReadGate) GetObjectNInfo(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions) (*GetObjectReader, error) {
+	r, err := g.ObjectLayer.GetObjectNInfo(ctx, bucket, object, rs, h, opts)
+	if err != nil {
+		return nil, err
+	}
+	mode, _ := ctx.Value(iamPolicyRefreshContext{}).(string)
+	var release <-chan struct{}
+	if bucket == minioMetaBucket && mode == "list" && object == g.policyPath {
+		release = g.listRelease
+	}
+	if bucket == minioMetaBucket && mode == "refresh" && object == g.userPath {
+		release = g.refreshRelease
+	}
+	if release == nil {
+		return r, nil
+	}
+	data, err := io.ReadAll(r)
+	info := r.ObjInfo
+	r.Close()
+	if err != nil {
+		return nil, err
+	}
+	if mode == "list" {
+		g.listOnce.Do(func() { close(g.listEntered) })
+	} else {
+		g.refreshOnce.Do(func() { close(g.refreshEntered) })
+	}
+	select {
+	case <-release:
+		return NewGetObjectReaderFromReader(bytes.NewReader(data), info, opts)
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+func TestIAMPolicyReloadActualObjectRefresh(t *testing.T) {
+	for _, docs := range []bool{false, true} {
+		for _, canceled := range []bool{false, true} {
+			t.Run(fmt.Sprintf("docs=%t/cancel=%t", docs, canceled), func(t *testing.T) {
+				ctx, sys, obj := prepareIAMRevisionFixture(t)
+				store := sys.store
+				rw, err := store.GetPolicy("readwrite")
+				mustIAM(t, err)
+				ro, err := store.GetPolicy("readonly")
+				mustIAM(t, err)
+				for _, name := range []string{"review-edit", "review-delete"} {
+					_, err = store.SetPolicy(ctx, name, rw)
+					mustIAM(t, err)
+				}
+				const parent = "review-parent"
+				_, err = sys.CreateUser(ctx, parent, madmin.AddOrUpdateUserReq{SecretKey: "synthetic-review-password", Status: madmin.AccountEnabled})
+				mustIAM(t, err)
+				secret, err := getTokenSigningKey()
+				mustIAM(t, err)
+				issue := func(typ string) auth.Credentials {
+					c, err := auth.GetNewCredentialsWithMetadata(map[string]any{"exp": UTCNow().Add(time.Hour).Unix(), parentClaim: parent, tokenRevokeTypeClaim: typ}, secret)
+					mustIAM(t, err)
+					c.ParentUser = parent
+					_, err = store.SetTempUser(ctx, c.AccessKey, c, "")
+					mustIAM(t, err)
+					return c
+				}
+				revoked, retained := issue("selected"), issue("other")
+				gate := &iamPolicyRefreshReadGate{ObjectLayer: obj, policyPath: getPolicyDocPath("review-edit"), userPath: getUserIdentityPath(parent, regUser), listEntered: make(chan struct{}), listRelease: make(chan struct{}), refreshEntered: make(chan struct{}), refreshRelease: make(chan struct{})}
+				store.IAMStorageAPI.(*IAMObjectStore).objAPI = gate
+				freeList := sync.OnceFunc(func() { close(gate.listRelease) })
+				freeRefresh := sync.OnceFunc(func() { close(gate.refreshRelease) })
+				var wg sync.WaitGroup
+				t.Cleanup(func() { freeList(); freeRefresh(); wg.Wait() })
+				start := func(fn func() error) <-chan error {
+					done := make(chan error, 1)
+					wg.Go(func() { done <- fn() })
+					return done
+				}
+				refreshDone := start(func() error {
+					return store.LoadIAMCache(context.WithValue(ctx, iamPolicyRefreshContext{}, "refresh"), false)
+				})
+				awaitIAMPolicyReload(t, gate.refreshEntered)
+				listCtx, cancel := context.WithCancel(context.WithValue(ctx, iamPolicyRefreshContext{}, "list"))
+				defer cancel()
+				listDone := start(func() error { return iamPolicyReloadCall(listCtx, store, docs) })
+				awaitIAMPolicyReload(t, gate.listEntered)
+				read := start(func() error {
+					if _, ok := store.GetUser(parent); !ok {
+						return errors.New("cached user lost")
+					}
+					_, err := store.GetPolicy("review-edit")
+					return err
+				})
+				select {
+				case err := <-read:
+					mustIAM(t, err)
+				case <-time.After(time.Second):
+					t.Fatal("actual object reload blocked cached reads")
+				}
+				writes := []<-chan error{
+					start(func() error { _, err := store.SetPolicy(ctx, "review-edit", ro); return err }),
+					start(func() error { return store.DeletePolicy(ctx, "review-delete", false) }),
+					start(func() error { _, err := store.RevokeTokens(ctx, parent, "selected"); return err }),
+				}
+				if canceled {
+					cancel()
+				} else {
+					freeList()
+				}
+				err = <-listDone
+				if canceled {
+					if !errors.Is(err, context.Canceled) {
+						t.Fatalf("cancel error: %v", err)
+					}
+				} else {
+					mustIAM(t, err)
+				}
+				for _, done := range writes {
+					mustIAM(t, <-done)
+				}
+				freeRefresh()
+				mustIAM(t, <-refreshDone)
+				p, err := store.GetPolicy("review-edit")
+				mustIAM(t, err)
+				if p.IsAllowed(policy.Args{Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) || !p.IsAllowed(policy.Args{Action: policy.GetObjectAction, BucketName: "bucket", ObjectName: "object"}) {
+					t.Fatal("actual object refresh restored stale policy")
+				}
+				if _, err := store.GetPolicy("review-delete"); !errors.Is(err, errNoSuchPolicy) {
+					t.Fatal("deleted policy restored")
+				}
+				if _, ok := store.GetUser(revoked.AccessKey); ok {
+					t.Fatal("revoked STS token restored")
+				}
+				if _, ok := store.GetUser(retained.AccessKey); !ok {
+					t.Fatal("unrelated token lost")
+				}
+			})
+		}
+	}
+}

cmd/iam-policy-reload_test.go

@@ -0,0 +1,373 @@
+// Copyright (c) 2026 Feng Ruohang
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/auth"
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+type iamPolicyReloadStore struct {
+	IAMStorageAPI
+	policies func(context.Context, map[string]PolicyDoc) error
+	users    func(context.Context, IAMUserType, map[string]UserIdentity) error
+}
+
+func (s *iamPolicyReloadStore) loadPolicyDocs(ctx context.Context, m map[string]PolicyDoc) error {
+	return s.policies(ctx, m)
+}
+
+func (s *iamPolicyReloadStore) loadUsers(ctx context.Context, typ IAMUserType, m map[string]UserIdentity) error {
+	if s.users != nil {
+		return s.users(ctx, typ, m)
+	}
+	return s.IAMStorageAPI.loadUsers(ctx, typ, m)
+}
+
+func iamPolicyReloadCall(ctx context.Context, store *IAMStoreSys, docs bool) error {
+	if docs {
+		_, err := store.ListPolicyDocs(ctx, "")
+		return err
+	}
+	_, err := store.ListPolicies(ctx, "")
+	return err
+}
+
+func awaitIAMPolicyReload(t *testing.T, ch <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(5 * time.Second):
+		t.Fatal("IAM operation did not reach the controlled barrier")
+	}
+}
+
+// Both list entry points must allow cached readers while preserving the writer
+// ordering used by policy edits, peer notifications and credential revocation.
+// A refresh that captured the old state must not publish after those writes.
+func TestIAMPolicyReloadConcurrentChanges(t *testing.T) {
+	for _, backend := range []string{"object", "etcd"} {
+		for _, docs := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/docs=%t", backend, docs), func(t *testing.T) {
+				ctx, sys, obj := prepareIAMRevisionFixture(t, backend)
+				store := sys.store
+				original := store.IAMStorageAPI
+				rw, err := store.GetPolicy("readwrite")
+				mustIAM(t, err)
+				ro, err := store.GetPolicy("readonly")
+				mustIAM(t, err)
+				for _, name := range []string{"reload-edit", "reload-delete", "reload-peer"} {
+					_, err := store.SetPolicy(ctx, name, rw)
+					mustIAM(t, err)
+				}
+				const parent = "reload-parent"
+				_, err = sys.CreateUser(ctx, parent, madmin.AddOrUpdateUserReq{SecretKey: "synthetic-reload-password", Status: madmin.AccountEnabled})
+				mustIAM(t, err)
+				secret, err := getTokenSigningKey()
+				mustIAM(t, err)
+				issue := func(typ string) auth.Credentials {
+					c, err := auth.GetNewCredentialsWithMetadata(map[string]any{"exp": UTCNow().Add(time.Hour).Unix(), parentClaim: parent, tokenRevokeTypeClaim: typ}, secret)
+					mustIAM(t, err)
+					c.ParentUser = parent
+					_, err = store.SetTempUser(ctx, c.AccessKey, c, "")
+					mustIAM(t, err)
+					return c
+				}
+				revoked, retained := issue("selected"), issue("other")
+				peer := &IAMStoreSys{IAMStorageAPI: newIAMObjectStore(obj, MinIOUsersSysType)}
+				if backend == "etcd" {
+					peer.IAMStorageAPI = newIAMEtcdStore(original.(*IAMEtcdStore).client, MinIOUsersSysType)
+				}
+				mustIAM(t, peer.LoadIAMCache(ctx, true))
+
+				type refreshKey struct{}
+				loading, releaseLoad := make(chan struct{}), make(chan struct{})
+				refreshing, releaseRefresh := make(chan struct{}), make(chan struct{})
+				unblockLoad := sync.OnceFunc(func() { close(releaseLoad) })
+				unblockRefresh := sync.OnceFunc(func() { close(releaseRefresh) })
+				var wg sync.WaitGroup
+				t.Cleanup(func() { unblockLoad(); unblockRefresh(); wg.Wait() })
+				store.IAMStorageAPI = &iamPolicyReloadStore{
+					IAMStorageAPI: original,
+					policies: func(ctx context.Context, m map[string]PolicyDoc) error {
+						if err := original.loadPolicyDocs(ctx, m); err != nil {
+							return err
+						}
+						if ctx.Value(refreshKey{}) == nil {
+							close(loading)
+							select {
+							case <-releaseLoad:
+							case <-ctx.Done():
+								return ctx.Err()
+							}
+						}
+						return nil
+					},
+					users: func(ctx context.Context, typ IAMUserType, m map[string]UserIdentity) error {
+						if err := original.loadUsers(ctx, typ, m); err != nil {
+							return err
+						}
+						if typ == stsUser && ctx.Value(refreshKey{}) != nil {
+							close(refreshing)
+							<-releaseRefresh
+						}
+						return nil
+					},
+				}
+				start := func(fn func() error) <-chan error {
+					out := make(chan error, 1)
+					wg.Go(func() { out <- fn() })
+					return out
+				}
+				refreshDone := start(func() error { return store.LoadIAMCache(context.WithValue(ctx, refreshKey{}, true), false) })
+				awaitIAMPolicyReload(t, refreshing)
+				listDone := start(func() error { return iamPolicyReloadCall(ctx, store, docs) })
+				awaitIAMPolicyReload(t, loading)
+				reader := start(func() error {
+					if _, ok := store.GetUser(parent); !ok {
+						return errors.New("committed user disappeared during policy loading")
+					}
+					_, err := store.GetPolicy("reload-edit")
+					return err
+				})
+				select {
+				case err := <-reader:
+					mustIAM(t, err)
+				case <-time.After(time.Second):
+					t.Fatal("policy loading blocked cached authentication")
+				}
+
+				// The peer commits independently; its notification waits alongside
+				// local changes while the older list snapshot remains in flight.
+				_, err = peer.SetPolicy(ctx, "reload-peer", ro)
+				mustIAM(t, err)
+				writersStarted := make(chan struct{}, 4)
+				startWrite := func(fn func() error) <-chan error {
+					return start(func() error { writersStarted <- struct{}{}; return fn() })
+				}
+				writes := []<-chan error{
+					startWrite(func() error { _, err := store.SetPolicy(ctx, "reload-edit", ro); return err }),
+					startWrite(func() error { return store.DeletePolicy(ctx, "reload-delete", false) }),
+					startWrite(func() error { _, err := store.RevokeTokens(ctx, parent, "selected"); return err }),
+					startWrite(func() error { return store.PolicyNotificationHandler(ctx, "reload-peer") }),
+				}
+				for range writes {
+					<-writersStarted
+				}
+				select {
+				case err := <-writes[0]:
+					t.Fatalf("policy write overtook the pending reload: %v", err)
+				case <-time.After(25 * time.Millisecond):
+				}
+				for _, done := range writes[1:] {
+					select {
+					case err := <-done:
+						t.Fatalf("IAM mutation overtook the pending reload: %v", err)
+					default:
+					}
+				}
+				unblockLoad()
+				mustIAM(t, <-listDone)
+				for _, done := range writes {
+					mustIAM(t, <-done)
+				}
+				unblockRefresh()
+				mustIAM(t, <-refreshDone)
+				for _, name := range []string{"reload-edit", "reload-peer"} {
+					p, err := store.GetPolicy(name)
+					mustIAM(t, err)
+					if p.IsAllowed(policy.Args{Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) {
+						t.Fatalf("stale list/refresh restored the old grant: %s", name)
+					}
+					if !p.IsAllowed(policy.Args{Action: policy.GetObjectAction, BucketName: "bucket", ObjectName: "object"}) {
+						t.Fatalf("reload lost the committed read grant: %s", name)
+					}
+				}
+				if _, err := store.GetPolicy("reload-delete"); !errors.Is(err, errNoSuchPolicy) {
+					t.Fatalf("deleted policy returned: %v", err)
+				}
+				if _, ok := store.GetUser(revoked.AccessKey); ok {
+					t.Fatal("stale list/refresh restored a revoked token")
+				}
+				for _, key := range []string{retained.AccessKey, parent} {
+					if _, ok := store.GetUser(key); !ok {
+						t.Fatal("reload removed an unrelated credential")
+					}
+				}
+			})
+		}
+	}
+}
+
+func TestIAMPolicyReloadFailurePreservesCache(t *testing.T) {
+	for _, docs := range []bool{false, true} {
+		for _, failure := range []string{"storage", "cancel_partial", "caller_deadline_partial"} {
+			t.Run(fmt.Sprintf("docs=%t/%s", docs, failure), func(t *testing.T) {
+				ctx, sys, _ := prepareIAMRevisionFixture(t)
+				var cancel context.CancelFunc
+				if failure == "caller_deadline_partial" {
+					ctx, cancel = context.WithTimeout(ctx, 100*time.Millisecond)
+				} else {
+					ctx, cancel = context.WithCancel(ctx)
+				}
+				defer cancel()
+				store := sys.store
+				cache := store.rlock()
+				updated := cache.updatedAt
+				store.runlock()
+				injected := errors.New("policy read failed")
+				entered := make(chan struct{})
+				original := store.IAMStorageAPI
+				store.IAMStorageAPI = &iamPolicyReloadStore{IAMStorageAPI: original, policies: func(ctx context.Context, m map[string]PolicyDoc) error {
+					m["partial-result"] = PolicyDoc{}
+					close(entered)
+					if failure == "storage" {
+						return injected
+					}
+					<-ctx.Done()
+					// A canceled enumeration can close without yielding an error.
+					return nil
+				}}
+				done := make(chan error, 1)
+				go func() { done <- iamPolicyReloadCall(ctx, store, docs) }()
+				awaitIAMPolicyReload(t, entered)
+				want := injected
+				switch failure {
+				case "cancel_partial":
+					cancel()
+					want = context.Canceled
+				case "caller_deadline_partial":
+					want = context.DeadlineExceeded
+				}
+				select {
+				case err := <-done:
+					if !errors.Is(err, want) {
+						t.Fatalf("got %v, want %v", err, want)
+					}
+				case <-time.After(time.Second):
+					cancel()
+					<-done
+					t.Fatal("reload did not honor caller cancellation")
+				}
+				cache = store.rlock()
+				_, partial := cache.iamPolicyDocsMap["partial-result"]
+				unchanged := cache.updatedAt.Equal(updated)
+				store.runlock()
+				if partial || !unchanged {
+					t.Fatal("failed/canceled reload published a partial policy map")
+				}
+				store.IAMStorageAPI = original
+				// Cancellation must release both locks so the next operation works.
+				mustIAM(t, iamPolicyReloadCall(context.Background(), store, docs))
+				if _, err := store.GetPolicy("readwrite"); err != nil {
+					t.Fatal("reload lost default policies")
+				}
+			})
+		}
+	}
+}
+
+type iamPolicyRelockObserver struct {
+	IAMStorageAPI
+	calls       atomic.Int32
+	reacquiring chan struct{}
+}
+
+func (s *iamPolicyRelockObserver) lock() *iamCache {
+	if s.calls.Add(1) == 2 {
+		close(s.reacquiring)
+	}
+	return s.IAMStorageAPI.lock()
+}
+
+func TestIAMPolicyReloadCallerBudgetPreserved(t *testing.T) {
+	for _, docs := range []bool{false, true} {
+		t.Run(fmt.Sprintf("docs=%t", docs), func(t *testing.T) {
+			_, sys, _ := prepareIAMRevisionFixture(t)
+			original := sys.store.IAMStorageAPI
+			prior := defaultContextTimeout
+			defaultContextTimeout = 50 * time.Millisecond
+			t.Cleanup(func() { defaultContextTimeout = prior })
+			sys.store.IAMStorageAPI = &iamPolicyReloadStore{IAMStorageAPI: original, policies: func(ctx context.Context, m map[string]PolicyDoc) error {
+				select {
+				case <-time.After(120 * time.Millisecond):
+					return original.loadPolicyDocs(ctx, m)
+				case <-ctx.Done():
+					return ctx.Err()
+				}
+			}}
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+			err := iamPolicyReloadCall(ctx, sys.store, docs)
+			if err != nil {
+				t.Fatalf("reload cut off a still-valid caller budget: err=%v callerErr=%v", err, ctx.Err())
+			}
+		})
+	}
+}
+
+func TestIAMPolicyReloadCancelWhileReacquiringCache(t *testing.T) {
+	for _, docs := range []bool{false, true} {
+		t.Run(fmt.Sprintf("docs=%t", docs), func(t *testing.T) {
+			_, sys, _ := prepareIAMRevisionFixture(t)
+			store := sys.store
+			original := store.IAMStorageAPI
+			observer := &iamPolicyRelockObserver{IAMStorageAPI: original, reacquiring: make(chan struct{})}
+			entered, release := make(chan struct{}), make(chan struct{})
+			store.IAMStorageAPI = &iamPolicyReloadStore{IAMStorageAPI: observer, policies: func(ctx context.Context, m map[string]PolicyDoc) error {
+				m["partial-reload"] = PolicyDoc{}
+				close(entered)
+				<-release
+				return nil
+			}}
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			done := make(chan error, 1)
+			go func() { done <- iamPolicyReloadCall(ctx, store, docs) }()
+			awaitIAMPolicyReload(t, entered)
+			cache := store.rlock()
+			before := cache.updatedAt
+			count := len(cache.iamPolicyDocsMap)
+			close(release)
+			awaitIAMPolicyReload(t, observer.reacquiring)
+			// This reader forces the reload to wait before its commit step.
+			// The request is canceled after I/O completed, before publishing.
+			cancel()
+			store.runlock()
+			err := <-done
+			if !errors.Is(err, context.Canceled) {
+				t.Fatalf("cancel before publish lost: %v", err)
+			}
+			cache = store.rlock()
+			_, partial := cache.iamPolicyDocsMap["partial-reload"]
+			unchanged := cache.updatedAt.Equal(before) && len(cache.iamPolicyDocsMap) == count
+			store.runlock()
+			if partial || !unchanged {
+				t.Fatal("canceled complete load still published")
+			}
+		})
+	}
+}

cmd/iam-store.go

@@ -1581,10 +1581,22 @@ func (store *IAMStoreSys) ListPolicies(ctx context.Context, bucketName string) (
 	defer store.unlock()
 
 	m := map[string]PolicyDoc{}
-	err := store.loadPolicyDocs(ctx, m)
+	// Keep cache writers serialized, but let authentication use the committed
+	// cache while the replacement is loaded into this private map.
+	err := func() error {
+		store.IAMStorageAPI.unlock()
+		defer store.IAMStorageAPI.lock()
+		if err := store.loadPolicyDocs(ctx, m); err != nil {
+			return err
+		}
+		return ctx.Err()
+	}()
 	if err != nil {
 		return nil, err
 	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 
 	// Sets default canned policies
 	setDefaultCannedPolicies(m)
@@ -1609,10 +1621,20 @@ func (store *IAMStoreSys) ListPolicyDocs(ctx context.Context, bucketName string)
 	defer store.unlock()
 
 	m := map[string]PolicyDoc{}
-	err := store.loadPolicyDocs(ctx, m)
+	err := func() error {
+		store.IAMStorageAPI.unlock()
+		defer store.IAMStorageAPI.lock()
+		if err := store.loadPolicyDocs(ctx, m); err != nil {
+			return err
+		}
+		return ctx.Err()
+	}()
 	if err != nil {
 		return nil, err
 	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 
 	// Sets default canned policies
 	setDefaultCannedPolicies(m)

cmd/kms-cancellation_test.go

@@ -0,0 +1,256 @@
+// Copyright (c) 2026 Feng Ruohang
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/minio/kms-go/kes"
+	"github.com/minio/minio/internal/config"
+	"github.com/minio/minio/internal/crypto"
+	"github.com/minio/minio/internal/kms"
+)
+
+// Use the configured KES client over TLS, rather than a mocked KMS interface:
+// cancellation must survive every layer, including the HTTP request context.
+func connectCancellationKES(t *testing.T, handler http.Handler) *kms.KMS {
+	t.Helper()
+	for _, key := range []string{kms.EnvKMSEndpoint, kms.EnvKMSSecretKey, kms.EnvKMSSecretKeyFile} {
+		t.Setenv(key, "")
+		mustIAM(t, os.Unsetenv(key))
+	}
+	server := httptest.NewTLSServer(handler)
+	t.Cleanup(server.Close)
+	ca := filepath.Join(t.TempDir(), "ca.pem")
+	mustIAM(t, os.WriteFile(ca, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw}), 0o600))
+	key, err := kes.GenerateAPIKey(nil)
+	mustIAM(t, err)
+	t.Setenv(kms.EnvKESEndpoint, server.URL)
+	t.Setenv(kms.EnvKESAPIKey, key.String())
+	t.Setenv(kms.EnvKESDefaultKey, "cancellation-key")
+	t.Setenv(kms.EnvKESServerCA, ca)
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	k, err := kms.Connect(ctx, nil)
+	mustIAM(t, err)
+	return k
+}
+
+func cancellationDecryptHandler(t *testing.T, k *kms.KMS, before func(*http.Request)) http.Handler {
+	t.Helper()
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// KMS Connect periodically describes the default key to keep it cached.
+		// This maintenance request is independent of the decrypt under test.
+		if r.Method == http.MethodGet && r.URL.Path == "/v1/key/describe/cancellation-key" {
+			w.Header().Set("Content-Type", "application/json")
+			_ = json.NewEncoder(w).Encode(map[string]string{"name": "cancellation-key"})
+			return
+		}
+		if r.URL.Path != "/v1/key/decrypt/cancellation-key" {
+			t.Errorf("unexpected KES path: %s", r.URL.Path)
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		var request struct {
+			Ciphertext []byte `json:"ciphertext"`
+			Context    []byte `json:"context"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
+			t.Error(err)
+			return
+		}
+		// Drain the HTTP/1 body so net/http can observe a peer disconnect while
+		// the handler is blocked in the simulated KES operation.
+		_, _ = io.Copy(io.Discard, r.Body)
+		_ = r.Body.Close()
+		before(r)
+		if r.Context().Err() != nil {
+			return
+		}
+		var aad kms.Context
+		if err := json.Unmarshal(request.Context, &aad); err != nil {
+			t.Error(err)
+			return
+		}
+		plain, err := k.Decrypt(r.Context(), &kms.DecryptRequest{Name: "cancellation-key", Ciphertext: request.Ciphertext, AssociatedData: aad})
+		w.Header().Set("Content-Type", "application/json")
+		if err != nil {
+			w.WriteHeader(http.StatusBadRequest)
+			_ = json.NewEncoder(w).Encode(map[string]string{"message": kes.ErrDecrypt.Error()})
+			return
+		}
+		_ = json.NewEncoder(w).Encode(map[string][]byte{"plaintext": plain})
+	})
+}
+
+func TestKMSRequestCancellation(t *testing.T) {
+	for _, operation := range []string{"iam", "sse-list"} {
+		for _, expiry := range []string{"cancel", "deadline"} {
+			t.Run(operation+"/"+expiry, func(t *testing.T) {
+				builtin, err := kms.NewBuiltin("cancellation-key", bytes.Repeat([]byte{42}, 32))
+				mustIAM(t, err)
+				entered, canceled, release := make(chan struct{}), make(chan struct{}), make(chan struct{})
+				var calls atomic.Int32
+				var enteredOnce, canceledOnce sync.Once
+				unblock := sync.OnceFunc(func() { close(release) })
+				client := connectCancellationKES(t, cancellationDecryptHandler(t, builtin, func(r *http.Request) {
+					calls.Add(1)
+					enteredOnce.Do(func() { close(entered) })
+					select {
+					case <-r.Context().Done():
+						canceledOnce.Do(func() { close(canceled) })
+					case <-release:
+					}
+				}))
+				t.Cleanup(unblock)
+				prior := GlobalKMS
+				GlobalKMS = client
+				t.Cleanup(func() { GlobalKMS = prior })
+				var run func(context.Context) error
+				if operation == "iam" {
+					const name = "config/iam/policies/cancel/policy.json"
+					plaintext := []byte(`{"Version":1}`)
+					ciphertext, err := config.EncryptBytes(builtin, plaintext, kms.Context{minioMetaBucket: minioMetaBucket + "/" + name})
+					mustIAM(t, err)
+					run = func(ctx context.Context) error {
+						got, err := decryptData(ctx, ciphertext, name)
+						if err == nil && !bytes.Equal(got, plaintext) {
+							return errors.New("config plaintext changed")
+						}
+						return err
+					}
+				} else {
+					var metadata []map[string]string
+					var buckets, names []string
+					var expected []crypto.ObjectKey
+					for i := range 3 {
+						name := fmt.Sprintf("object-%d", i)
+						dek, err := builtin.GenerateKey(context.Background(), &kms.GenerateKeyRequest{AssociatedData: kms.Context{"bucket": "bucket/" + name}})
+						mustIAM(t, err)
+						key := crypto.GenerateKey(dek.Plaintext, nil)
+						sealed := key.Seal(dek.Plaintext, crypto.GenerateIV(nil), crypto.S3.String(), "bucket", name)
+						metadata = append(metadata, crypto.S3.CreateMetadata(nil, dek.KeyID, dek.Ciphertext, sealed))
+						buckets, names = append(buckets, "bucket"), append(names, name)
+						expected = append(expected, key)
+					}
+					run = func(ctx context.Context) error {
+						keys, err := crypto.S3.UnsealObjectKeys(ctx, client, metadata, buckets, names)
+						if err != nil {
+							return err
+						}
+						if len(keys) != len(expected) {
+							return errors.New("partial key list returned as success")
+						}
+						for i := range keys {
+							if keys[i] != expected[i] {
+								return errors.New("object key or ordering changed")
+							}
+						}
+						return nil
+					}
+				}
+				ctx, cancel := context.WithCancel(context.Background())
+				if expiry == "deadline" {
+					cancel()
+					ctx, cancel = context.WithTimeout(context.Background(), 300*time.Millisecond)
+				}
+				defer cancel()
+				done := make(chan error, 1)
+				go func() { done <- run(ctx) }()
+				select {
+				case <-entered:
+				case err := <-done:
+					t.Fatalf("decrypt returned before reaching KES: %v (type=%v)", err, client.Type)
+				case <-time.After(5 * time.Second):
+					t.Fatal("decrypt did not reach KES")
+				}
+				want := context.DeadlineExceeded
+				if expiry == "cancel" {
+					cancel()
+					want = context.Canceled
+				}
+				select {
+				case err := <-done:
+					if !errors.Is(err, want) {
+						t.Fatalf("got %v, want %v", err, want)
+					}
+				// The current KES SDK may sleep twice for up to 999ms after a
+				// deadline error. Include that existing backoff plus the 300ms
+				// request deadline; cancellation must still stop new decrypts.
+				case <-time.After(3 * time.Second):
+					unblock()
+					<-done
+					t.Fatal("request exceeded the cancellation and KES retry budget")
+				}
+				awaitIAMPolicyReload(t, canceled)
+				if n := calls.Load(); n != 1 {
+					t.Fatalf("cancellation launched a fallback or more object decryptions: %d calls", n)
+				}
+				unblock()
+				mustIAM(t, run(context.Background()))
+				before := calls.Load()
+				cctx, ccancel := context.WithCancel(context.Background())
+				ccancel()
+				if err := run(cctx); !errors.Is(err, context.Canceled) || calls.Load() != before {
+					t.Fatal("already-canceled request still issued decryption work")
+				}
+			})
+		}
+	}
+}
+
+func TestIAMDecryptAssociatedDataCompatibility(t *testing.T) {
+	builtin, err := kms.NewBuiltin("cancellation-key", bytes.Repeat([]byte{42}, 32))
+	mustIAM(t, err)
+	var calls atomic.Int32
+	client := connectCancellationKES(t, cancellationDecryptHandler(t, builtin, func(*http.Request) { calls.Add(1) }))
+	prior := GlobalKMS
+	GlobalKMS = client
+	t.Cleanup(func() { GlobalKMS = prior })
+	const name = "config/iam/policies/compat/policy.json"
+	plaintext := []byte(`{"Version":1}`)
+	for _, legacy := range []bool{false, true} {
+		t.Run(fmt.Sprint(legacy), func(t *testing.T) {
+			aad, wantCalls := minioMetaBucket+"/"+name, int32(1)
+			if legacy {
+				aad, wantCalls = name, 2
+			}
+			ciphertext, err := config.EncryptBytes(builtin, plaintext, kms.Context{minioMetaBucket: aad})
+			mustIAM(t, err)
+			before := calls.Load()
+			got, err := decryptData(context.Background(), ciphertext, name)
+			mustIAM(t, err)
+			if !bytes.Equal(got, plaintext) || calls.Load()-before != wantCalls {
+				t.Fatal("current/legacy encrypted IAM records no longer decrypt correctly")
+			}
+		})
+	}
+}

cmd/sftp-server-driver_test.go

@@ -0,0 +1,187 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/pkg/sftp"
+	"golang.org/x/crypto/ssh"
+)
+
+// Exercise the production RequestServer driver against a real S3 backend.
+// The library's filesystem Server and its ReadOnly option are a different path.
+func TestSFTPRequestServerObjectOperations(t *testing.T) {
+	for _, secure := range []bool{false, true} {
+		t.Run(fmt.Sprintf("TLS=%t", secure), func(t *testing.T) {
+			c := &check{t, "ErasureSD"}
+			suite := newTestSuiteIAM(TestSuiteCommon{serverType: "ErasureSD", signer: signerV4, secure: secure}, false)
+			suite.SetUpSuite(c)
+			t.Cleanup(func() { suite.TearDownSuite(c) })
+			previousTLS, previousTransport := globalIsTLS, globalRemoteFTPClientTransport
+			globalIsTLS = secure
+			globalRemoteFTPClientTransport = suite.TestSuiteCommon.client.Transport.(*http.Transport)
+			t.Cleanup(func() {
+				globalIsTLS, globalRemoteFTPClientTransport = previousTLS, previousTransport
+			})
+
+			ctx, cancel := context.WithTimeout(context.Background(), testDefaultTimeout)
+			defer cancel()
+			connect := func(policy string) *sftp.Client {
+				t.Helper()
+				accessKey, secretKey := mustGenerateCredentials(c)
+				if err := suite.adm.SetUser(ctx, accessKey, secretKey, madmin.AccountEnabled); err != nil {
+					t.Fatal(err)
+				}
+				if _, err := suite.adm.AttachPolicy(ctx, madmin.PolicyAssociationReq{Policies: []string{policy}, User: accessKey}); err != nil {
+					t.Fatal(err)
+				}
+				perms, err := sshPasswordAuth(newSSHConnMock(accessKey), []byte(secretKey))
+				if err != nil {
+					t.Fatal(err)
+				}
+				return newSFTPRequestTestClient(t, perms)
+			}
+			client := connect("readwrite")
+			bucket := "/" + getRandomBucketName()
+			if err := client.Mkdir(bucket); err != nil {
+				t.Fatal(err)
+			}
+			read := func(client *sftp.Client, name string) []byte {
+				t.Helper()
+				f, err := client.Open(name)
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer f.Close()
+				data, err := io.ReadAll(f)
+				if err != nil {
+					t.Fatal(err)
+				}
+				return data
+			}
+			for _, size := range []int{0, 32767, 32768, 65537} {
+				name := fmt.Sprintf("%s/中文-café-%d", bucket, size)
+				data := bytes.Repeat([]byte{byte(size)}, size)
+				f, err := client.Create(name)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if n, err := f.ReadFrom(bytes.NewReader(data)); err != nil || n != int64(size) {
+					f.Close()
+					t.Fatalf("upload %s: n=%d, err=%v", name, n, err)
+				}
+				if err := f.Close(); err != nil {
+					t.Fatal(err)
+				}
+				if got := read(client, name); !bytes.Equal(got, data) {
+					t.Fatalf("download %s: byte mismatch", name)
+				}
+				info, err := client.Stat(name)
+				if err != nil || info.Size() != int64(size) {
+					t.Fatalf("stat %s: info=%v, err=%v", name, info, err)
+				}
+			}
+			entries, err := client.ReadDir(bucket)
+			if err != nil || len(entries) != 4 {
+				t.Fatalf("list: %d entries, err=%v", len(entries), err)
+			}
+			name := bucket + "/中文-café-65537"
+			original := read(client, name)
+			// Include list permission because Client.Remove also tries Rmdir;
+			// both requests must observe the denied delete, not a failed listing.
+			readonly := connect("consolereadonly")
+			if !bytes.Equal(read(readonly, name), original) {
+				t.Fatal("readonly download changed content")
+			}
+			if f, err := readonly.OpenFile(name, os.O_WRONLY|os.O_TRUNC); err == nil {
+				_, writeErr := f.Write([]byte("forbidden"))
+				closeErr := f.Close()
+				// The streaming driver can acknowledge buffered bytes before S3
+				// rejects the PUT. Verify the object, not just the write status.
+				t.Logf("readonly write: write=%v close=%v", writeErr, closeErr)
+			}
+			if !bytes.Equal(read(client, name), original) {
+				t.Fatal("readonly upload modified content")
+			}
+			if err := readonly.Remove(name); err == nil {
+				t.Fatal("readonly delete succeeded")
+			}
+			if !bytes.Equal(read(client, name), original) {
+				t.Fatal("denied operation modified content")
+			}
+			// Remove retries Rmdir, which can mask a listing error for a user
+			// without ListBucket. Check the stored object even if it returns nil.
+			withoutList := connect("readonly")
+			withoutList.Remove(name)
+			if !bytes.Equal(read(client, name), original) {
+				t.Fatal("readonly user without listing permission deleted content")
+			}
+			for operation, err := range map[string]error{
+				"hardlink": client.Link(name, bucket+"/hardlink"),
+				"symlink":  client.Symlink(name, bucket+"/symlink"),
+				"rename":   client.Rename(name, bucket+"/renamed"),
+			} {
+				if status, ok := err.(*sftp.StatusError); !ok || status.Code != uint32(sftp.ErrSSHFxOpUnsupported) {
+					t.Fatalf("%s: expected unsupported operation, got %v", operation, err)
+				}
+			}
+			for _, entry := range entries {
+				if err := client.Remove(bucket + "/" + entry.Name()); err != nil {
+					t.Fatal(err)
+				}
+			}
+			entries, err = client.ReadDir(bucket)
+			if err != nil || len(entries) != 0 {
+				t.Fatalf("list after deletion: %d entries, err=%v", len(entries), err)
+			}
+			// SILO exposes a missing object as a directory for stat-then-list
+			// clients. Keep that existing virtual-directory behavior.
+			if info, err := client.Stat(name); err != nil || !info.IsDir() {
+				t.Fatalf("stat virtual directory: info=%v, err=%v", info, err)
+			}
+			if err := client.RemoveDirectory(bucket); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+func newSFTPRequestTestClient(t *testing.T, perms *ssh.Permissions) *sftp.Client {
+	t.Helper()
+	serverConn, clientConn := net.Pipe()
+	deadline := time.Now().Add(time.Minute)
+	serverConn.SetDeadline(deadline)
+	clientConn.SetDeadline(deadline)
+	server := sftp.NewRequestServer(serverConn, NewSFTPDriver(perms, "127.0.0.1"), sftp.WithRSAllocator())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		server.Serve()
+	}()
+	t.Cleanup(func() {
+		clientConn.Close()
+		server.Close()
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Error("SFTP request server did not stop")
+		}
+	})
+	client, err := sftp.NewClientPipe(clientConn, clientConn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { client.Close() })
+	return client
+}