From 2616d43fbc3400a02cc1e6f9299e26ae8b2e71a6 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Fri, 16 Oct 2026 21:49:38 +0800 Subject: [PATCH 1/2] fix(iam): enforce admin credential authorization boundaries Honor self-service authorization denials, prevent session-limited credential delegation and updates, and protect root credential mutations under built-in IAM. Preserve empty-policy inheritance and damaged-account cleanup, normalize new service policy claims at issuance, and add signed Admin API and LDAP regressions. Signed-off-by: Feng Ruohang --- cmd/admin-handlers-idp-ldap.go | 4 +- cmd/admin-handlers-users-authz_test.go | 507 +++++++++++++++++++++++++ cmd/admin-handlers-users.go | 221 +++++++---- cmd/iam-peer-reload_test.go | 2 + cmd/sts-handlers_test.go | 22 +- 5 files changed, 671 insertions(+), 85 deletions(-) create mode 100644 cmd/admin-handlers-users-authz_test.go diff --git a/cmd/admin-handlers-idp-ldap.go b/cmd/admin-handlers-idp-ldap.go index 0695f45f6..415605f1b 100644 --- a/cmd/admin-handlers-idp-ldap.go +++ b/cmd/admin-handlers-idp-ldap.go @@ -249,8 +249,10 @@ func (a adminAPIHandlers) AddServiceAccountLDAP(w http.ResponseWriter, r *http.R // In case of LDAP/OIDC we need to set `opts.claims` to ensure // it is associated with the LDAP/OIDC user properly. + // The new credential's policy comes from the request, not the caller. for k, v := range cred.Claims { - if k == expClaim { + switch k { + case expClaim, policy.SessionPolicyName, sessionPolicyNameExtracted, iamPolicyClaimNameSA(): continue } opts.claims[k] = v diff --git a/cmd/admin-handlers-users-authz_test.go b/cmd/admin-handlers-users-authz_test.go new file mode 100644 index 000000000..4ee840eb7 --- /dev/null +++ b/cmd/admin-handlers-users-authz_test.go @@ -0,0 +1,507 @@ +// Copyright (c) 2026 PGSTY +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/minio/madmin-go/v3" + "github.com/minio/minio/internal/auth" + polplugin "github.com/minio/minio/internal/config/policy/plugin" + "github.com/minio/mux" + xnet "github.com/pgsty/silo-pkg/v3/net" + "github.com/pgsty/silo-pkg/v3/policy" +) + +func iamAdminFixture(t *testing.T) (context.Context, *IAMSys) { + t.Helper() + ctx, sys, obj := prepareIAMRevisionFixture(t) + globalObjLayerMutex.Lock() + globalObjectAPI = obj + globalObjLayerMutex.Unlock() + return ctx, sys +} + +func iamAdminUser(ctx context.Context, t *testing.T, sys *IAMSys, name, policyName string) auth.Credentials { + t.Helper() + cred := auth.Credentials{AccessKey: name, SecretKey: "review-user-valid-secret"} + _, err := sys.CreateUser(ctx, name, madmin.AddOrUpdateUserReq{SecretKey: cred.SecretKey, Status: madmin.AccountEnabled}) + mustIAM(t, err) + if policyName != "" { + _, err = sys.PolicyDBSet(ctx, name, policyName, regUser, false) + mustIAM(t, err) + } + return cred +} + +func iamAdminServiceAccount(ctx context.Context, t *testing.T, sys *IAMSys, parent, key, document string) auth.Credentials { + t.Helper() + var sp *policy.Policy + var err error + if document != "" { + sp, err = policy.ParseConfig(strings.NewReader(document)) + mustIAM(t, err) + } + cred, _, err := sys.NewServiceAccount(ctx, parent, nil, newServiceAccountOpts{accessKey: key, secretKey: "review-service-valid-secret", sessionPolicy: sp}) + mustIAM(t, err) + return cred +} + +func iamAdminRequest(t *testing.T, cred auth.Credentials, method, target string, payload any) *http.Request { + t.Helper() + var body []byte + if payload != nil { + raw, err := json.Marshal(payload) + mustIAM(t, err) + body, err = madmin.EncryptData(cred.SecretKey, raw) + mustIAM(t, err) + } + headers := map[string]string{} + if cred.IsTemp() && !cred.IsServiceAccount() { + headers["X-Amz-Security-Token"] = cred.SessionToken + } + req, err := newTestSignedRequestV4(method, "http://localhost"+target, int64(len(body)), bytes.NewReader(body), cred.AccessKey, cred.SecretKey, headers) + mustIAM(t, err) + mustIAM(t, req.ParseForm()) + return req +} + +func iamAdminCall(t *testing.T, cred auth.Credentials, method, target string, payload any, vars map[string]string, handler http.HandlerFunc) *httptest.ResponseRecorder { + t.Helper() + req := iamAdminRequest(t, cred, method, target, payload) + req = mux.SetURLVars(req, vars) + rec := httptest.NewRecorder() + req = req.WithContext(newContext(req, rec, "IAMAdminAuthorization")) + handler(rec, req) + return rec +} + +func assertIAMAdminStatus(t *testing.T, operation string, rec *httptest.ResponseRecorder, want int) { + t.Helper() + if rec.Code != want { + t.Errorf("%s status=%d want=%d body=%s", operation, rec.Code, want, rec.Body.String()) + return + } + if want == http.StatusForbidden || want == http.StatusNotFound { + var response APIErrorResponse + mustIAM(t, json.Unmarshal(rec.Body.Bytes(), &response)) + wantCode := errorCodes.ToAPIErr(ErrAccessDenied).Code + if want == http.StatusNotFound { + wantCode = errorCodes.ToAPIErr(ErrAdminServiceAccountNotFound).Code + } + if response.Code != wantCode { + t.Errorf("%s error code=%s want=%s", operation, response.Code, wantCode) + } + } +} + +func iamAdminPlugin(t *testing.T, allow bool) { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]bool{"result": allow}) + })) + t.Cleanup(server.Close) + u, err := xnet.ParseHTTPURL(server.URL) + mustIAM(t, err) + previous := newGlobalAuthZPluginFn() + setGlobalAuthZPlugin(polplugin.New(polplugin.Args{URL: u, CloseRespFn: func(r io.ReadCloser) { r.Close() }})) + t.Cleanup(func() { setGlobalAuthZPlugin(previous) }) +} + +func TestIAMAdminRootEmptyPolicy(t *testing.T) { + for _, doc := range []string{"", "null", "{}", `{"Statement":null}`} { + t.Run("policy="+doc, func(t *testing.T) { + ctx, sys := iamAdminFixture(t) + caller := iamAdminServiceAccount(ctx, t, sys, globalActiveCred.AccessKey, "review-root-empty", doc) + sibling := iamAdminServiceAccount(ctx, t, sys, globalActiveCred.AccessKey, "review-root-sibling", "") + req := iamAdminRequest(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil) + checked, owner, code := checkKeyValid(req, caller.AccessKey) + if code != ErrNone { + t.Fatalf("checkKeyValid: %v", code) + } + active, _ := isAllowedBySessionPolicyForServiceAccount(policy.Args{Claims: checked.Claims}) + t.Logf("owner=%v parsed_active_session_policy=%v", owner, active) + api := adminAPIHandlers{} + assertIAMAdminStatus(t, "create", iamAdminCall(t, caller, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{AccessKey: "review-root-new", SecretKey: "review-service-valid-secret"}, nil, api.AddServiceAccount), 200) + assertIAMAdminStatus(t, "update", iamAdminCall(t, caller, "POST", "/minio/admin/v3/update-service-account?accessKey="+sibling.AccessKey, madmin.UpdateServiceAccountReq{NewName: "review"}, map[string]string{"accessKey": sibling.AccessKey}, api.UpdateServiceAccount), 204) + assertIAMAdminStatus(t, "delete", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+sibling.AccessKey, nil, map[string]string{"accessKey": sibling.AccessKey}, api.DeleteServiceAccount), 204) + assertIAMAdminStatus(t, "revoke", iamAdminCall(t, caller, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true", nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 204) + }) + } +} + +func TestIAMAdminEmptyPolicyDelegation(t *testing.T) { + for _, parentType := range []string{"root", "user"} { + for _, document := range []string{"null", "{}", `{"Statement":null}`} { + t.Run(parentType+"/"+document, func(t *testing.T) { + ctx, sys := iamAdminFixture(t) + parent := globalActiveCred + if parentType == "user" { + parent = iamAdminUser(ctx, t, sys, "empty-policy-parent", "readwrite") + } + sp, err := policy.ParseConfig(strings.NewReader(document)) + mustIAM(t, err) + caller, _, err := sys.NewServiceAccount(ctx, parent.AccessKey, nil, newServiceAccountOpts{ + accessKey: "empty-policy-caller", secretKey: "empty-policy-caller-secret", sessionPolicy: sp, + claims: map[string]any{"identity-attribute": "preserved"}, + }) + mustIAM(t, err) + const limitedPolicy = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:ListServiceAccounts"]}]}` + for _, tc := range []struct { + policy string + listStatus int + allowObject bool + }{ + {"", http.StatusOK, true}, + {"null", http.StatusOK, true}, + {"{}", http.StatusOK, true}, + {`{"Statement":null}`, http.StatusOK, true}, + {limitedPolicy, http.StatusOK, false}, + {`{"Version":"2012-10-17"}`, http.StatusForbidden, false}, + {`{"Statement":[]}`, http.StatusForbidden, false}, + } { + expires := UTCNow().Add(time.Hour) + req := madmin.AddServiceAccountReq{Expiration: &expires} + if tc.policy != "" { + req.Policy = json.RawMessage(tc.policy) + } + api := adminAPIHandlers{} + rec := iamAdminCall(t, caller, "PUT", "/minio/admin/v3/add-service-account", req, nil, api.AddServiceAccount) + assertIAMAdminStatus(t, "create expiring child", rec, http.StatusOK) + data, err := madmin.DecryptData(caller.SecretKey, rec.Body) + mustIAM(t, err) + var response madmin.AddServiceAccountResp + mustIAM(t, json.Unmarshal(data, &response)) + child, ok := sys.store.GetUser(response.Credentials.AccessKey) + if !ok { + t.Fatal("new service account missing") + } + // Exercise the issued key through authentication and authorization. + assertIAMAdminStatus(t, "use expiring child policy="+tc.policy, iamAdminCall(t, child.Credentials, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), tc.listStatus) + signed := iamAdminRequest(t, child.Credentials, "GET", "/example/object", nil) + checked, owner, code := checkKeyValid(signed, child.Credentials.AccessKey) + if code != ErrNone { + t.Fatalf("child authentication: %v", code) + } + if checked.Claims["identity-attribute"] != "preserved" { + t.Fatal("creation lost the caller's identity context") + } + allowed := sys.IsAllowed(policy.Args{AccountName: checked.AccessKey, Claims: checked.Claims, IsOwner: owner, Action: policy.GetObjectAction, BucketName: "example", ObjectName: "object"}) + if allowed != tc.allowObject { + t.Errorf("GetObject allowed=%v with requested policy=%s", allowed, tc.policy) + } + } + }) + } + } +} + +func TestIAMAdminCorruptServiceCleanup(t *testing.T) { + for _, status := range []string{auth.AccountOn, auth.AccountOff, "expired"} { + t.Run(status, func(t *testing.T) { + ctx, sys := iamAdminFixture(t) + caller := iamAdminUser(ctx, t, sys, "review-admin", "consoleAdmin") + parent := iamAdminUser(ctx, t, sys, "review-parent", "readwrite") + svc := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "rv-corrupt", "") + if status == auth.AccountOff { + _, err := sys.UpdateServiceAccount(ctx, svc.AccessKey, updateServiceAccountOpts{status: status}) + mustIAM(t, err) + } + cache := sys.store.lock() + u := cache.iamUsersMap[svc.AccessKey] + u.Credentials.SessionToken = "broken-session-token" + if status == "expired" { + u.Credentials.Expiration = UTCNow().Add(-time.Minute) + } + cache.iamUsersMap[svc.AccessKey] = u + sys.store.unlock() + _, _, err := sys.GetServiceAccount(ctx, svc.AccessKey) + if err == nil { + t.Fatal("fault injection failed") + } + raw, exists := sys.store.GetUser(svc.AccessKey) + _, loginOK := sys.GetUser(ctx, svc.AccessKey) + t.Logf("metadata_exists=%v parent_known=%v GetUser_ok=%v", exists, raw.Credentials.ParentUser == parent.AccessKey, loginOK) + api := adminAPIHandlers{} + outsider := iamAdminUser(ctx, t, sys, "cleanup-outsider", "readwrite") + assertIAMAdminStatus(t, "unauthorized cleanup", iamAdminCall(t, outsider, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.DeleteServiceAccount), 404) + assertIAMAdminStatus(t, "admin cleanup", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.DeleteServiceAccount), 204) + if _, ok := sys.store.GetUser(svc.AccessKey); ok { + t.Fatal("deleted service account remains in the IAM cache") + } + }) + } +} + +func TestIAMAdminPolicylessSelf(t *testing.T) { + ctx, sys := iamAdminFixture(t) + caller := iamAdminUser(ctx, t, sys, "review-policyless", "") + svc := iamAdminServiceAccount(ctx, t, sys, caller.AccessKey, "rv-policyless-child", "") + api := adminAPIHandlers{} + assertIAMAdminStatus(t, "list", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), 200) + assertIAMAdminStatus(t, "info service", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.InfoServiceAccount), 200) + assertIAMAdminStatus(t, "info access key", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-access-key?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.InfoAccessKey), 200) + assertIAMAdminStatus(t, "delete", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.DeleteServiceAccount), 204) + assertIAMAdminStatus(t, "revoke", iamAdminCall(t, caller, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true", nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 204) + assertIAMAdminStatus(t, "list another user", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts?user="+globalActiveCred.AccessKey, nil, nil, api.ListServiceAccounts), 403) + assertIAMAdminStatus(t, "revoke another user", iamAdminCall(t, caller, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true&user="+globalActiveCred.AccessKey, nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 403) +} + +func TestIAMAdminExpiringInheritedPolicy(t *testing.T) { + ctx, sys := iamAdminFixture(t) + parent := iamAdminUser(ctx, t, sys, "review-legacy-parent", "consoleAdmin") + document := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:CreateServiceAccount","admin:UpdateServiceAccount"]}]}` + // Model a credential persisted by the old handler: it copied its caller's + // policy claims but set the newly created service's type to inherited. + expires := UTCNow().Add(time.Hour) + caller, _, err := sys.NewServiceAccount(ctx, parent.AccessKey, nil, newServiceAccountOpts{ + accessKey: "rv-exp-inherited", secretKey: "review-service-valid-secret", expiration: &expires, + claims: map[string]any{policy.SessionPolicyName: base64.StdEncoding.EncodeToString([]byte(document))}, + }) + mustIAM(t, err) + req := iamAdminRequest(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil) + checked, owner, code := checkKeyValid(req, caller.AccessKey) + if code != ErrNone { + t.Fatalf("checkKeyValid: %v", code) + } + args := policy.Args{AccountName: caller.AccessKey, Claims: checked.Claims, IsOwner: owner, Action: policy.CreateServiceAccountAdminAction, DenyOnly: true} + if !sys.IsAllowed(args) { + t.Fatal("fixture cannot create") + } + args.Action = policy.CreateUserAdminAction + args.DenyOnly = false + if sys.IsAllowed(args) { + t.Fatal("fixture not constrained") + } + t.Logf("service=%v temp=%v inherited=%v; create allowed, create-user denied", checked.IsServiceAccount(), checked.IsTemp(), checked.Claims[iamPolicyClaimNameSA()] == inheritedPolicyType) + api := adminAPIHandlers{} + rec := iamAdminCall(t, caller, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{AccessKey: "review-escaped", SecretKey: "review-service-valid-secret"}, nil, api.AddServiceAccount) + if rec.Code == 200 { + child, ok := sys.GetUser(ctx, "review-escaped") + if !ok { + t.Fatal("new key missing") + } + childReq := iamAdminRequest(t, child.Credentials, "GET", "/minio/admin/v3/list-service-accounts", nil) + childChecked, childOwner, childCode := checkKeyValid(childReq, child.Credentials.AccessKey) + if childCode != ErrNone { + t.Fatalf("child auth=%v", childCode) + } + allowed := sys.IsAllowed(policy.Args{AccountName: childChecked.AccessKey, Claims: childChecked.Claims, IsOwner: childOwner, Action: policy.CreateUserAdminAction}) + t.Errorf("constrained inherited service created a new key; new key CreateUser allowed=%v", allowed) + } else { + assertIAMAdminStatus(t, "reject derived creation", rec, 403) + } + updated := iamAdminCall(t, caller, "POST", "/minio/admin/v3/update-service-account?accessKey="+caller.AccessKey, madmin.UpdateServiceAccountReq{NewName: "metadata-only"}, map[string]string{"accessKey": caller.AccessKey}, api.UpdateServiceAccount) + if updated.Code == 204 { + after, ok := sys.GetUser(ctx, caller.AccessKey) + if !ok { + t.Fatal("updated key missing") + } + afterReq := iamAdminRequest(t, after.Credentials, "GET", "/minio/admin/v3/list-service-accounts", nil) + afterChecked, afterOwner, afterCode := checkKeyValid(afterReq, caller.AccessKey) + if afterCode != ErrNone { + t.Fatalf("updated auth=%v", afterCode) + } + allowed := sys.IsAllowed(policy.Args{AccountName: caller.AccessKey, Claims: afterChecked.Claims, IsOwner: afterOwner, Action: policy.CreateUserAdminAction}) + t.Errorf("constrained inherited service updated itself; CreateUser allowed after update=%v", allowed) + } else { + assertIAMAdminStatus(t, "reject inherited self update", updated, 403) + } +} + +func TestIAMAdminSessionPolicyCannotDelegate(t *testing.T) { + for _, temp := range []bool{false, true} { + t.Run(map[bool]string{false: "service", true: "sts"}[temp], func(t *testing.T) { + ctx, sys := iamAdminFixture(t) + parent := iamAdminUser(ctx, t, sys, "rv-parent", "consoleAdmin") + doc := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:CreateServiceAccount","admin:UpdateServiceAccount"]}]}` + var caller auth.Credentials + if !temp { + caller = iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "review-restricted", doc) + } else { + var err error + caller, err = auth.GetNewCredentialsWithMetadata(map[string]any{parentClaim: parent.AccessKey, "exp": UTCNow().Add(time.Hour).Unix(), policy.SessionPolicyName: base64.StdEncoding.EncodeToString([]byte(doc))}, globalActiveCred.SecretKey) + mustIAM(t, err) + caller.ParentUser = parent.AccessKey + _, err = sys.SetTempUser(ctx, caller.AccessKey, caller, "") + mustIAM(t, err) + } + sibling := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "rv-sibling", "") + api := adminAPIHandlers{} + assertIAMAdminStatus(t, "restrict create", iamAdminCall(t, caller, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{AccessKey: "review-bad-new", SecretKey: "review-service-valid-secret"}, nil, api.AddServiceAccount), 403) + assertIAMAdminStatus(t, "restrict LDAP create", iamAdminCall(t, caller, "PUT", "/minio/admin/v3/idp/ldap/add-service-account", madmin.AddServiceAccountReq{AccessKey: "review-bad-new", SecretKey: "review-service-valid-secret"}, nil, api.AddServiceAccountLDAP), 403) + assertIAMAdminStatus(t, "restrict update", iamAdminCall(t, caller, "POST", "/minio/admin/v3/update-service-account?accessKey="+sibling.AccessKey, madmin.UpdateServiceAccountReq{NewSecretKey: "review-known-new-secret"}, map[string]string{"accessKey": sibling.AccessKey}, api.UpdateServiceAccount), 403) + if _, ok := sys.store.GetUser("review-bad-new"); ok { + t.Fatal("denied creation persisted a credential") + } + u, ok := sys.store.GetUser(sibling.AccessKey) + if !ok || u.Credentials.SecretKey != sibling.SecretKey { + t.Fatal("denied update changed the target credential") + } + }) + } +} + +func TestIAMAdminSelfDenials(t *testing.T) { + for _, mode := range []string{"parent_deny", "group_deny", "session_implicit_deny", "session_explicit_deny", "plugin_deny"} { + t.Run(mode, func(t *testing.T) { + ctx, sys := iamAdminFixture(t) + parent := iamAdminUser(ctx, t, sys, "rv-self-parent", "readwrite") + target := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "rv-self-target", "") + caller := parent + doc := `{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":["admin:ListServiceAccounts","admin:RemoveServiceAccount"]}]}` + switch mode { + case "parent_deny", "group_deny": + p, err := policy.ParseConfig(strings.NewReader(doc)) + mustIAM(t, err) + _, err = sys.SetPolicy(ctx, "review-self-deny", *p) + mustIAM(t, err) + if mode == "group_deny" { + _, err = sys.AddUsersToGroup(ctx, "self-deny-group", []string{parent.AccessKey}) + mustIAM(t, err) + _, err = sys.PolicyDBSet(ctx, "self-deny-group", "review-self-deny", regUser, true) + } else { + _, err = sys.PolicyDBSet(ctx, parent.AccessKey, "readwrite,review-self-deny", regUser, false) + } + mustIAM(t, err) + case "session_implicit_deny": + caller = iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "rv-self-caller", `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::example/*"]}]}`) + case "session_explicit_deny": + caller = iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "rv-self-caller", doc) + case "plugin_deny": + iamAdminPlugin(t, false) + } + api := adminAPIHandlers{} + assertIAMAdminStatus(t, "list", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), 403) + assertIAMAdminStatus(t, "info service", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.InfoServiceAccount), 403) + assertIAMAdminStatus(t, "info access key", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-access-key?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.InfoAccessKey), 403) + assertIAMAdminStatus(t, "delete", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.DeleteServiceAccount), 403) + assertIAMAdminStatus(t, "revoke", iamAdminCall(t, caller, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true", nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 403) + }) + } +} + +func TestIAMAdminNonRootCannotManageRoot(t *testing.T) { + ctx, sys := iamAdminFixture(t) + caller := iamAdminUser(ctx, t, sys, "rv-console-admin", "consoleAdmin") + target := iamAdminServiceAccount(ctx, t, sys, globalActiveCred.AccessKey, "rv-root-target", "") + api := adminAPIHandlers{} + assertIAMAdminStatus(t, "create root", iamAdminCall(t, caller, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{AccessKey: "rv-new-root", SecretKey: "review-service-valid-secret", TargetUser: globalActiveCred.AccessKey}, nil, api.AddServiceAccount), 403) + assertIAMAdminStatus(t, "update root", iamAdminCall(t, caller, "POST", "/minio/admin/v3/update-service-account?accessKey="+target.AccessKey, madmin.UpdateServiceAccountReq{NewSecretKey: "review-known-new-secret"}, map[string]string{"accessKey": target.AccessKey}, api.UpdateServiceAccount), 403) + assertIAMAdminStatus(t, "delete root", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.DeleteServiceAccount), 404) + assertIAMAdminStatus(t, "revoke root", iamAdminCall(t, caller, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true&user="+globalActiveCred.AccessKey, nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 403) + // Restricted root-derived credentials and the site replicator are not root + // administrators, even when their effective policy allows these actions. + limited := iamAdminServiceAccount(ctx, t, sys, globalActiveCred.AccessKey, "limited-root", `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:*"]}]}`) + assertIAMAdminStatus(t, "limited root delete", iamAdminCall(t, limited, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.DeleteServiceAccount), 404) + assertIAMAdminStatus(t, "limited root revoke", iamAdminCall(t, limited, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true", nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 403) + replicator, _, err := sys.NewServiceAccount(ctx, globalActiveCred.AccessKey, nil, newServiceAccountOpts{accessKey: siteReplicatorSvcAcc, secretKey: "replicator-valid-secret", allowSiteReplicatorAccount: true}) + mustIAM(t, err) + assertIAMAdminStatus(t, "replicator create root", iamAdminCall(t, replicator, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{}, nil, api.AddServiceAccount), 403) + + // An external plugin remains authoritative, including for root targets. + iamAdminPlugin(t, true) + assertIAMAdminStatus(t, "plugin update root", iamAdminCall(t, limited, "POST", "/minio/admin/v3/update-service-account?accessKey="+target.AccessKey, madmin.UpdateServiceAccountReq{NewName: "plugin-authorized"}, map[string]string{"accessKey": target.AccessKey}, api.UpdateServiceAccount), 204) + assertIAMAdminStatus(t, "plugin create root", iamAdminCall(t, limited, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{}, nil, api.AddServiceAccount), 200) +} + +func TestIAMAdminDeleteCorruptServiceReplicates(t *testing.T) { + ctx, sys := iamAdminFixture(t) + caller := iamAdminUser(ctx, t, sys, "cleanup-admin", "consoleAdmin") + parent := iamAdminUser(ctx, t, sys, "cleanup-parent", "readwrite") + target := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "cleanup-target", "") + cache := sys.store.lock() + u := cache.iamUsersMap[target.AccessKey] + u.Credentials.SessionToken = "broken-session-token" + u.Credentials.Status = auth.AccountOff + cache.iamUsersMap[target.AccessKey] = u + sys.store.unlock() + + var deletions atomic.Int32 + peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/minio/health/live" { + return + } + var batch iamRevisionBatch + if err := json.NewDecoder(r.Body).Decode(&batch); err != nil { + t.Error(err) + w.WriteHeader(http.StatusBadRequest) + return + } + if len(batch.Items) != 1 || batch.Items[0].SvcAccChange == nil || batch.Items[0].SvcAccChange.Delete == nil || batch.Items[0].SvcAccChange.Delete.AccessKey != target.AccessKey { + t.Error("replication did not carry the service account deletion") + } else { + deletions.Add(1) + } + _ = json.NewEncoder(w).Encode(iamRevisionResponse{iamRevisionStatus: iamRevisionStatus{Version: iamRevisionProtocol, Node: "peer", Instance: "peer-boot", Digest: "deleted"}}) + })) + t.Cleanup(peer.Close) + globalSiteReplicatorCred.RLock() + previousSecret := globalSiteReplicatorCred.secretKey + globalSiteReplicatorCred.RUnlock() + globalSiteReplicatorCred.Set(caller.SecretKey) + t.Cleanup(func() { globalSiteReplicatorCred.Set(previousSecret) }) + globalSiteReplicationSys.Lock() + globalSiteReplicationSys.enabled = true + globalSiteReplicationSys.state = srState{ServiceAccountAccessKey: caller.AccessKey, Peers: map[string]madmin.PeerInfo{ + globalDeploymentID(): {DeploymentID: globalDeploymentID(), Name: "local"}, + "peer": {DeploymentID: "peer", Name: "peer", Endpoint: peer.URL}, + }} + globalSiteReplicationSys.Unlock() + t.Cleanup(func() { + globalSiteReplicationSys.Lock() + globalSiteReplicationSys.enabled = false + globalSiteReplicationSys.Unlock() + }) + api := adminAPIHandlers{} + assertIAMAdminStatus(t, "replicated cleanup", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.DeleteServiceAccount), 204) + if deletions.Load() != 1 { + t.Fatalf("expected one immediate deletion notification, got %d", deletions.Load()) + } +} + +func TestIAMAdminListSelfAliases(t *testing.T) { + for _, kind := range []string{"service", "sts"} { + t.Run(kind, func(t *testing.T) { + ctx, sys := iamAdminFixture(t) + parent := iamAdminUser(ctx, t, sys, "alias-parent", "readwrite") + target := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-target", "") + caller := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-caller", "") + if kind == "sts" { + var err error + caller, err = auth.GetNewCredentialsWithMetadata(map[string]any{parentClaim: parent.AccessKey, "exp": UTCNow().Add(time.Hour).Unix()}, globalActiveCred.SecretKey) + mustIAM(t, err) + caller.ParentUser = parent.AccessKey + _, err = sys.SetTempUser(ctx, caller.AccessKey, caller, "") + mustIAM(t, err) + } + api := adminAPIHandlers{} + for _, alias := range []string{"", caller.AccessKey, parent.AccessKey} { + rec := iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts?user="+alias, nil, nil, api.ListServiceAccounts) + assertIAMAdminStatus(t, "list self alias", rec, 200) + data, err := madmin.DecryptData(caller.SecretKey, rec.Body) + mustIAM(t, err) + var response madmin.ListServiceAccountsResp + mustIAM(t, json.Unmarshal(data, &response)) + found := false + for _, account := range response.Accounts { + found = found || account.AccessKey == target.AccessKey + } + if !found { + t.Fatalf("self alias %q did not list the parent's service accounts", alias) + } + } + }) + } +} diff --git a/cmd/admin-handlers-users.go b/cmd/admin-handlers-users.go index 77eb24f52..848a00d6d 100644 --- a/cmd/admin-handlers-users.go +++ b/cmd/admin-handlers-users.go @@ -714,8 +714,10 @@ func (a adminAPIHandlers) AddServiceAccount(w http.ResponseWriter, r *http.Reque // In case of LDAP/OIDC we need to set `opts.claims` to ensure // it is associated with the LDAP/OIDC user properly. + // The new credential's policy comes from the request, not the caller. for k, v := range cred.Claims { - if k == expClaim { + switch k { + case expClaim, policy.SessionPolicyName, sessionPolicyNameExtracted, iamPolicyClaimNameSA(): continue } opts.claims[k] = v @@ -855,10 +857,9 @@ func (a adminAPIHandlers) UpdateServiceAccount(w http.ResponseWriter, r *http.Re // Permission checks: // - // 1. Any type of account (i.e. access keys (previously/still called service - // accounts), STS accounts, internal IDP accounts, etc) with the - // policy.UpdateServiceAccountAdminAction permission can update any service - // account. + // 1. Updating requires policy.UpdateServiceAccountAdminAction. Under built-in + // IAM, session-limited callers cannot update credentials, and only unrestricted + // root credentials can manage root's service accounts. // // 2. We would like to let a user update their own access keys, however it // is currently blocked pending a re-design. Users are still able to delete @@ -874,6 +875,11 @@ func (a adminAPIHandlers) UpdateServiceAccount(w http.ResponseWriter, r *http.Re writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) return } + if newGlobalAuthZPluginFn() == nil && (hasActiveSessionPolicy(cred) || + (svcAccount.ParentUser == globalActiveCred.AccessKey && !canManageRootCredentials(cred))) { + writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) + return + } var sp *policy.Policy if len(updateReq.NewPolicy) > 0 { @@ -951,23 +957,21 @@ func (a adminAPIHandlers) InfoServiceAccount(w http.ResponseWriter, r *http.Requ return } - if !globalIAMSys.IsAllowed(policy.Args{ + requestUser := cred.AccessKey + if cred.ParentUser != "" { + requestUser = cred.ParentUser + } + if !isAllowedServiceAccountAction(cred, policy.Args{ AccountName: cred.AccessKey, Groups: cred.Groups, Action: policy.ListServiceAccountsAdminAction, ConditionValues: getConditionValues(r, "", cred), IsOwner: owner, Claims: cred.Claims, + DenyOnly: requestUser == svcAccount.ParentUser, }) { - requestUser := cred.AccessKey - if cred.ParentUser != "" { - requestUser = cred.ParentUser - } - - if requestUser != svcAccount.ParentUser { - writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) - return - } + writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) + return } // if session policy is nil or empty, then it is implied policy @@ -1039,29 +1043,27 @@ func (a adminAPIHandlers) ListServiceAccounts(w http.ResponseWriter, r *http.Req return } - var targetAccount string - - // If listing is requested for a specific user (who is not the request - // sender), check that the user has permissions. + selfUser := cred.AccessKey + if cred.ParentUser != "" { + selfUser = cred.ParentUser + } user := r.Form.Get("user") - if user != "" && user != cred.AccessKey { - if !globalIAMSys.IsAllowed(policy.Args{ - AccountName: cred.AccessKey, - Groups: cred.Groups, - Action: policy.ListServiceAccountsAdminAction, - ConditionValues: getConditionValues(r, "", cred), - IsOwner: owner, - Claims: cred.Claims, - }) { - writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) - return - } - targetAccount = user - } else { - targetAccount = cred.AccessKey - if cred.ParentUser != "" { - targetAccount = cred.ParentUser - } + selfOnly := user == "" || user == cred.AccessKey || user == selfUser + if !isAllowedServiceAccountAction(cred, policy.Args{ + AccountName: cred.AccessKey, + Groups: cred.Groups, + Action: policy.ListServiceAccountsAdminAction, + ConditionValues: getConditionValues(r, "", cred), + IsOwner: owner, + Claims: cred.Claims, + DenyOnly: selfOnly, + }) { + writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) + return + } + targetAccount := user + if selfOnly { + targetAccount = selfUser } serviceAccounts, err := globalIAMSys.ListServiceAccounts(ctx, targetAccount) @@ -1110,7 +1112,7 @@ func (a adminAPIHandlers) DeleteServiceAccount(w http.ResponseWriter, r *http.Re // Get current object layer instance. objectAPI := newObjectLayerFn() - if objectAPI == nil || globalNotificationSys == nil { + if objectAPI == nil || globalNotificationSys == nil || !globalIAMSys.Initialized() { writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL) return } @@ -1131,35 +1133,41 @@ func (a adminAPIHandlers) DeleteServiceAccount(w http.ResponseWriter, r *http.Re writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidArgument), r.URL) return } - // We do not care if service account is readable or not at this point, - // since this is a delete call we shall allow it to be deleted if possible. - svcAccount, _, err := globalIAMSys.GetServiceAccount(ctx, serviceAccount) - if errors.Is(err, errNoSuchServiceAccount) { + // Deletion needs the stored identity, not a valid token or session policy. + // Keep disabled, expired and damaged service accounts removable, and retain + // their parent for authorization and the site replication hook below. + u, ok := globalIAMSys.store.GetUser(serviceAccount) + if !ok || !u.Credentials.IsServiceAccount() { writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminServiceAccountNotFound), r.URL) return } + svcAccount := u.Credentials - adminPrivilege := globalIAMSys.IsAllowed(policy.Args{ + parentUser := cred.AccessKey + if cred.ParentUser != "" { + parentUser = cred.ParentUser + } + isSelf := parentUser == svcAccount.ParentUser + if !isAllowedServiceAccountAction(cred, policy.Args{ AccountName: cred.AccessKey, Groups: cred.Groups, Action: policy.RemoveServiceAccountAdminAction, ConditionValues: getConditionValues(r, "", cred), IsOwner: owner, Claims: cred.Claims, - }) - - if !adminPrivilege { - parentUser := cred.AccessKey - if cred.ParentUser != "" { - parentUser = cred.ParentUser - } - if svcAccount.ParentUser != "" && parentUser != svcAccount.ParentUser { - // The service account belongs to another user but return not - // found error to mitigate brute force attacks. or the - // serviceAccount doesn't exist. + DenyOnly: isSelf, + }) { + if !isSelf { + // Hide the existence of another user's service account. writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminServiceAccountNotFound), r.URL) return } + writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) + return + } + if newGlobalAuthZPluginFn() == nil && svcAccount.ParentUser == globalActiveCred.AccessKey && !canManageRootCredentials(cred) { + writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminServiceAccountNotFound), r.URL) + return } if err := globalIAMSys.DeleteServiceAccount(ctx, serviceAccount, true); err != nil { @@ -2046,26 +2054,30 @@ func (a adminAPIHandlers) RevokeTokens(w http.ResponseWriter, r *http.Request) { return } - adminPrivilege := globalIAMSys.IsAllowed(policy.Args{ + parentUser := cred.AccessKey + if cred.ParentUser != "" { + parentUser = cred.ParentUser + } + isSelfTarget := isTokenSelfRevoke || user == parentUser + if !isAllowedServiceAccountAction(cred, policy.Args{ AccountName: cred.AccessKey, Groups: cred.Groups, Action: policy.RemoveServiceAccountAdminAction, ConditionValues: getConditionValues(r, "", cred), IsOwner: owner, Claims: cred.Claims, - }) - - if !adminPrivilege || isTokenSelfRevoke { - parentUser := cred.AccessKey - if cred.ParentUser != "" { - parentUser = cred.ParentUser - } - if !isTokenSelfRevoke && user != parentUser { - writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) - return - } + DenyOnly: isSelfTarget, + }) { + writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) + return + } + if isSelfTarget { user = parentUser } + if newGlobalAuthZPluginFn() == nil && user == globalActiveCred.AccessKey && !canManageRootCredentials(cred) { + writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) + return + } // Infer token revoke type from the request if requestor is STS. if isTokenSelfRevoke && tokenRevokeType == "" && !fullRevoke { @@ -2112,29 +2124,22 @@ func (a adminAPIHandlers) InfoAccessKey(w http.ResponseWriter, r *http.Request) u, ok := globalIAMSys.GetUser(ctx, accessKey) targetCred := u.Credentials - if !globalIAMSys.IsAllowed(policy.Args{ + requestUser := cred.AccessKey + if cred.ParentUser != "" { + requestUser = cred.ParentUser + } + isSelf := ok && targetCred.ParentUser != "" && requestUser == targetCred.ParentUser + if !isAllowedServiceAccountAction(cred, policy.Args{ AccountName: cred.AccessKey, Groups: cred.Groups, Action: policy.ListServiceAccountsAdminAction, ConditionValues: getConditionValues(r, "", cred), IsOwner: owner, Claims: cred.Claims, + DenyOnly: isSelf, }) { - // If requested user does not exist and requestor is not allowed to list service accounts, return access denied. - if !ok { - writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) - return - } - - requestUser := cred.AccessKey - if cred.ParentUser != "" { - requestUser = cred.ParentUser - } - - if requestUser != targetCred.ParentUser { - writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) - return - } + writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL) + return } if !ok { @@ -2908,6 +2913,44 @@ func addExpirationToCondValues(exp *time.Time, condValues map[string][]string) e return nil } +// hasActiveSessionPolicy follows the same credential dispatch as IAMSys.IsAllowed. +// Expiring service accounts also satisfy IsTemp and are evaluated as STS there. +// Callers use this only after authorization under built-in IAM. +func hasActiveSessionPolicy(cred auth.Credentials) bool { + args := policy.Args{Claims: cred.Claims} + if cred.IsTemp() { + hasSP, _ := isAllowedBySessionPolicy(args) + return hasSP + } + if cred.IsServiceAccount() { + if cred.Claims[iamPolicyClaimNameSA()] == inheritedPolicyType { + return false + } + hasSP, _ := isAllowedBySessionPolicyForServiceAccount(args) + return hasSP + } + return false +} + +// canManageRootCredentials is an additional guard after built-in IAM authorization. +// The signature's owner flag is insufficient: historical empty session policies +// clear it even though the service account still inherits root's permissions. +func canManageRootCredentials(cred auth.Credentials) bool { + return cred.AccessKey == globalActiveCred.AccessKey || + (cred.ParentUser == globalActiveCred.AccessKey && cred.AccessKey != siteReplicatorSvcAcc && !hasActiveSessionPolicy(cred)) +} + +// isAllowedServiceAccountAction preserves regular users' self inspection and +// cleanup even without a policy attachment, while enforcing explicit denies. +// Derived credentials and external authorization plugins use full IAM evaluation. +func isAllowedServiceAccountAction(cred auth.Credentials, args policy.Args) bool { + if args.DenyOnly && !args.IsOwner && !cred.IsTemp() && !cred.IsServiceAccount() && newGlobalAuthZPluginFn() == nil { + policies, err := globalIAMSys.PolicyDBGet(args.AccountName, args.Groups...) + return err == nil && globalIAMSys.GetCombinedPolicy(policies...).IsAllowed(args) + } + return globalIAMSys.IsAllowed(args) +} + func commonAddServiceAccount(r *http.Request, ldap bool) (context.Context, auth.Credentials, newServiceAccountOpts, madmin.AddServiceAccountReq, string, APIError) { ctx := r.Context() @@ -2996,6 +3039,14 @@ func commonAddServiceAccount(r *http.Request, ldap bool) (context.Context, auth. }) { return ctx, auth.Credentials{}, newServiceAccountOpts{}, madmin.AddServiceAccountReq{}, "", errorCodes.ToAPIErr(ErrAccessDenied) } + effectiveTarget := targetUser + if (targetUser == cred.AccessKey || targetUser == cred.ParentUser) && cred.ParentUser != "" { + effectiveTarget = cred.ParentUser + } + if newGlobalAuthZPluginFn() == nil && (hasActiveSessionPolicy(cred) || + (effectiveTarget == globalActiveCred.AccessKey && !canManageRootCredentials(cred))) { + return ctx, auth.Credentials{}, newServiceAccountOpts{}, madmin.AddServiceAccountReq{}, "", errorCodes.ToAPIErr(ErrAccessDenied) + } var sp *policy.Policy if len(createReq.Policy) > 0 { @@ -3003,6 +3054,12 @@ func commonAddServiceAccount(r *http.Request, ldap bool) (context.Context, auth. if err != nil { return ctx, auth.Credentials{}, newServiceAccountOpts{}, madmin.AddServiceAccountReq{}, "", toAdminAPIErr(ctx, err) } + // Empty service policies mean inheritance, including for expiring keys. + // Normalize the replication request as well as the local credential. + if sp.Version == "" && sp.Statements == nil && sp.ID == "" { + sp = nil + createReq.Policy = nil + } } opts.sessionPolicy = sp diff --git a/cmd/iam-peer-reload_test.go b/cmd/iam-peer-reload_test.go index b1d09be6e..181d84b99 100644 --- a/cmd/iam-peer-reload_test.go +++ b/cmd/iam-peer-reload_test.go @@ -24,6 +24,8 @@ func TestIAMPeerDeleteNotificationReloadsCommittedState(t *testing.T) { resetTestGlobals() ctx, cancel := context.WithCancel(context.Background()) defer cancel() + // Both caches use object storage, independent of earlier etcd suites. + globalEtcdClient = nil obj, disk, err := prepareFS(ctx) if err != nil { t.Fatal(err) diff --git a/cmd/sts-handlers_test.go b/cmd/sts-handlers_test.go index 214188a14..dc58dcc69 100644 --- a/cmd/sts-handlers_test.go +++ b/cmd/sts-handlers_test.go @@ -44,6 +44,7 @@ import ( "github.com/minio/minio-go/v7/pkg/tags" xhttp "github.com/minio/minio/internal/http" "github.com/pgsty/silo-pkg/v3/ldap" + "github.com/pgsty/silo-pkg/v3/policy" ) func runAllIAMSTSTests(suite *TestSuiteIAM, c *check) { @@ -2384,8 +2385,10 @@ func (s *TestSuiteIAM) TestIAMImport(c *check, exportedContent []byte, caseNum i if err != nil { c.Fatalf("import %d: Unable to list policies: %v", caseNum, err) } - defaultCannedPolicies := set.CreateStringSet("consoleAdmin", "readwrite", "readonly", - "diagnostics", "writeonly") + defaultCannedPolicies := set.CreateStringSet() + for _, p := range policy.DefaultPolicies { + defaultCannedPolicies.Add(p.Name) + } for policy, policyBytes := range policyContentMap { if defaultCannedPolicies.Contains(policy) { continue @@ -2991,6 +2994,21 @@ func (s *TestSuiteIAM) TestLDAPSTSServiceAccounts(c *check) { // 6. Check that service account cannot be created for some other user. c.mustNotCreateSvcAccount(ctx, globalActiveCred.AccessKey, userAdmClient) + // An empty policy must not contaminate an expiring LDAP child credential. + emptyPolicyCred, err := userAdmClient.AddServiceAccountLDAP(ctx, madmin.AddServiceAccountReq{Policy: []byte("null")}) + if err != nil { + c.Fatalf("Unable to create empty-policy LDAP service account: %v", err) + } + emptyPolicyAdmin := s.getAdminClient(c, emptyPolicyCred.AccessKey, emptyPolicyCred.SecretKey, "") + expires := UTCNow().Add(time.Hour) + for _, childPolicy := range [][]byte{nil, []byte("null")} { + child, err := emptyPolicyAdmin.AddServiceAccountLDAP(ctx, madmin.AddServiceAccountReq{Expiration: &expires, Policy: childPolicy}) + if err != nil { + c.Fatalf("Unable to create expiring LDAP child: %v", err) + } + c.assertSvcAccS3Access(ctx, s, child, bucket) + } + // Detach the policy from the user if _, err = s.adm.DetachPolicyLDAP(ctx, userReq); err != nil { c.Fatalf("Unable to detach user policy: %v", err) From 4ed86dfe62ed6e07016885552feebcb59f779631 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 17 Oct 2026 08:56:50 +0800 Subject: [PATCH 2/2] Unify durable multipart listing without a migration mode Signed-off-by: Feng Ruohang --- CHANGELOG.md | 52 ++-- cmd/api-errors.go | 4 +- cmd/common-main.go | 4 + ...rasure-multipart-listing-benchmark_test.go | 85 ++++++ cmd/erasure-multipart-listing.go | 53 ++-- cmd/erasure-multipart-listing_test.go | 122 +++------ cmd/erasure-multipart-migration_test.go | 255 ++++++----------- cmd/erasure-multipart-unified_test.go | 256 ++++++++++++++++++ cmd/erasure-multipart.go | 82 +----- cmd/erasure-server-pool.go | 76 +----- cmd/erasure-sets.go | 15 +- cmd/handler-api.go | 8 - cmd/list-multipart-uploads-compat_test.go | 30 +- internal/config/api/api.go | 11 +- internal/config/api/api_test.go | 35 ++- 15 files changed, 589 insertions(+), 499 deletions(-) create mode 100644 cmd/erasure-multipart-listing-benchmark_test.go create mode 100644 cmd/erasure-multipart-unified_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 36bb3ecd4..98bc7e05e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,35 +54,29 @@ and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-0 ### Object storage and replication -- Make `ListMultipartUploads` discover quorum-valid uploads from durable state - across pools, erasure sets and drives, then apply S3 prefix, delimiter, - marker, ordering and 1,000-entry pagination semantics globally (#198). New uploads - store their canonical bucket and key as reserved fields in the existing - quorum-written `xl.meta`; completion removes those upload-only fields. Native - markers remain usable after their upload is completed or canceled. Strict - listing returns a diagnostic 503 for legacy uploads or uncertain coverage; - the default remains the released exact-key/cache-based `legacy` behavior. - Opt into strict mode only through `MINIO_API_MULTIPART_LISTING=strict`, after - upgrading every writer, draining old uploads, checking the read-only admin - `multipart-preflight` report and validating scan capacity. The process-only - setting is not persisted into shared API configuration. Historical - `multipart_listing` keys are ignored and can be removed with a targeted - `mcli admin config reset ALIAS api multipart_listing` before rollback. Per-process - admission, directory-entry, worker and time budgets bound scan scheduling; - each page still scans durable state. See [issue #79](https://github.com/pgsty/silo/issues/79) - and its [design record](https://silo.pgsty.com/blog/design/list-multipart-uploads/). - Thanks to mr javad seydi (@mrjavadseydi) for the original implementation. -- Retain released read-quorum and best-effort multipart cancellation in default - legacy mode. Strict mode requires majority deletion acknowledgements and - permits retries below read quorum. When most drives were already empty, - failed deletion of an observed remnant now returns 503 instead of being - masked by empty-drive successes. Wrong-key or wrong-bucket cancellation - preserves the valid upload's cache entry; successful cancellation notifies - peers with the request context after the distributed lock is released. - The HTTP response for an absent - upload remains 204; this is not proof of physical cleanup. **Known boundary:** - delayed creation writes can still restore an upload after cancellation; - this change does not add a durable creation fence. +- Make durable `ListMultipartUploads` scanning the only listing path (#79): + apply prefix, delimiter, markers, ordering and the 1,000-entry page limit + globally across pools and sets, independently of node-local caches. Omit + readable historical uploads only when both bucket/key identity fields are + absent; malformed identities, uncertain coverage and resource limits remain + errors. Old uploads still support part upload, completion and cancellation + with a known bucket/key/upload ID. No backfill or new storage layout is added. + `MINIO_API_MULTIPART_LISTING` is retired and ignored with a startup warning; + historical shared `api multipart_listing` keys remain tolerated. The read-only + admin `multipart-preflight` reports `mode=persistent`; `ready` diagnoses the + absence of old records and no longer gates listing. Each page still scans + durable state within the existing 100,000-directory-entry, 30-second and + per-process concurrency budgets. See the + [design and limits](https://silo.pgsty.com/blog/design/list-multipart-uploads/#implementation). + Thanks to mr javad seydi (@mrjavadseydi) for the original metadata-and-scan + implementation (#198). +- Keep the default read-quorum, best-effort multipart cancellation and + first-matching-pool behavior when retiring the listing switch. The optional + strict cancellation branch is removed. Wrong-target protection, upload-ID + path validation and peer notification with the request context remain. + HTTP 204 for an absent upload remains idempotent and does not prove physical + cleanup. **Known boundary:** delayed creation writes can still restore a + readable upload after cancellation; this change adds no creation fence. - Preserve object tags during multi-pool metadata reconciliation by reading the resolved tag field together with its revision (#189). Previously, reconciliation could replace existing tags with an empty value. diff --git a/cmd/api-errors.go b/cmd/api-errors.go index 7de44a39a..ef3665c2d 100644 --- a/cmd/api-errors.go +++ b/cmd/api-errors.go @@ -450,7 +450,7 @@ const ( ErrAdminNoSecretKey ErrIAMNotInitialized - ErrMultipartListingLegacy + ErrMultipartListingLegacy // Reserved: legacy uploads no longer block listing. ErrMultipartListingIdentity ErrSlowDown @@ -2191,8 +2191,6 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) { err = unwrapAll(err) switch err { - case errMultipartListingLegacy: - apiErr = ErrMultipartListingLegacy case errMultipartListingIdentity: apiErr = ErrMultipartListingIdentity case errCompleteMultipartChecksumMismatch, errCompleteMultipartChecksumTypeMismatch: diff --git a/cmd/common-main.go b/cmd/common-main.go index c127759fc..47bd895ab 100644 --- a/cmd/common-main.go +++ b/cmd/common-main.go @@ -56,6 +56,7 @@ import ( "github.com/minio/minio/internal/auth" "github.com/minio/minio/internal/color" "github.com/minio/minio/internal/config" + "github.com/minio/minio/internal/config/api" "github.com/minio/minio/internal/handlers" "github.com/minio/minio/internal/kms" "github.com/minio/minio/internal/logger" @@ -881,6 +882,9 @@ func serverHandleEnvVars() { logger.Warning("%s=%s is ignored: Silo in-place updates are permanently disabled", config.EnvUpdate, updateSetting) } globalInplaceUpdateDisabled = true + if env.IsSet(api.EnvAPIMultipartListing) { + logger.Warning("%s is ignored: multipart listing always uses persistent metadata; uploads without identity fields are omitted", api.EnvAPIMultipartListing) + } // Check if the supported credential env vars, // "MINIO_ROOT_USER" and "MINIO_ROOT_PASSWORD" are provided diff --git a/cmd/erasure-multipart-listing-benchmark_test.go b/cmd/erasure-multipart-listing-benchmark_test.go new file mode 100644 index 000000000..7dd430a5b --- /dev/null +++ b/cmd/erasure-multipart-listing-benchmark_test.go @@ -0,0 +1,85 @@ +// Copyright (c) 2026 Ruohang Feng +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "testing" +) + +// Run explicitly with -benchtime=1x. Fixture creation is excluded from timing; +// each request still discovers real on-disk metadata. The 16-drive/10k case +// deliberately exceeds the existing aggregate directory-entry budget. +func BenchmarkMultipartListingCapacity(b *testing.B) { + for _, tc := range []struct { + drives, uploads int + mixed, limited bool + }{ + {4, 1_000, false, false}, + {4, 10_000, false, false}, + {4, 10_000, true, false}, + {16, 1_000, false, false}, + {16, 10_000, false, true}, + } { + b.Run(fmt.Sprintf("drives=%d/uploads=%d/mixed=%t", tc.drives, tc.uploads, tc.mixed), func(b *testing.B) { + obj, dirs, err := prepareErasure(b.Context(), tc.drives) + if err != nil { + b.Fatal(err) + } + z := obj.(*erasureServerPools) + b.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) }) + const bucket, other = "multipart-capacity", "multipart-unrelated" + for _, name := range []string{bucket, other} { + if err := z.MakeBucket(b.Context(), name, MakeBucketOptions{}); err != nil { + b.Fatal(err) + } + } + for n := range tc.uploads { + target := bucket + if tc.mixed && n%2 != 0 { + target = other + } + key := fmt.Sprintf("dir/%05d", n) + mp, err := z.NewMultipartUpload(b.Context(), target, key, ObjectOptions{}) + if err != nil { + b.Fatal(err) + } + if tc.mixed && n%10 == 0 { + rewriteMultipartIdentity(b, z.serverPools[0].getHashedSet(key), target, key, mp.UploadID, removeMultipartIdentity) + } + } + z.mpCache.Clear() + var directoryCalls, metadataCalls atomic.Int64 + for _, set := range z.serverPools[0].sets { + original := set.getDisks + set.getDisks = func() []StorageAPI { + disks := append([]StorageAPI(nil), original()...) + for i, disk := range disks { + disks[i] = multipartListingCountingDisk{disk, &directoryCalls, &metadataCalls} + } + return disks + } + b.Cleanup(func() { set.getDisks = original }) + } + b.ResetTimer() + for range b.N { + got, err := z.ListMultipartUploads(b.Context(), bucket, "dir/", "", "", "", 1000) + if tc.limited { + var limited SlowDown + if !errors.As(err, &limited) || len(got.Uploads) != 0 { + b.Fatalf("overflow must return SlowDown without a partial page: %+v %v", got, err) + } + } else if err != nil || len(got.Uploads) != 1000 { + b.Fatalf("capacity listing: count=%d err=%v", len(got.Uploads), err) + } + } + b.StopTimer() + b.ReportMetric(float64(directoryCalls.Load())/float64(b.N), "ListDir/op") + b.ReportMetric(float64(metadataCalls.Load())/float64(b.N), "ReadVersion/op") + }) + } +} diff --git a/cmd/erasure-multipart-listing.go b/cmd/erasure-multipart-listing.go index cbc9b15fa..252a8ef6b 100644 --- a/cmd/erasure-multipart-listing.go +++ b/cmd/erasure-multipart-listing.go @@ -23,7 +23,6 @@ const multipartScanEntryLimit = 100_000 var ( multipartScanSlots = make(chan struct{}, 2) - errMultipartListingLegacy = errors.New("legacy multipart uploads require a coordinated upgrade and drain") errMultipartListingIdentity = errors.New("multipart upload identity is invalid") ) @@ -133,6 +132,14 @@ func (er erasureObjects) multipartIdentity(fi FileInfo, shaDir string) (string, IsValidObjectPrefix(object) && er.getMultipartSHADir(bucket, object) == shaDir } +// Only the absence of both fields identifies the historical format. Empty, +// partial or invalid identities must not hide a damaged new upload as legacy. +func multipartLegacyIdentity(fi FileInfo) bool { + _, bucketPresent := fi.Metadata[multipartMetaBucket] + _, objectPresent := fi.Metadata[multipartMetaObject] + return !bucketPresent && !objectPresent +} + func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket, candidate string, source StorageAPI) (MultipartInfo, bool, bool, error) { if s.ctx.Err() != nil { return MultipartInfo{}, false, false, SlowDown{} @@ -162,8 +169,8 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket, disks := er.getDisks() metadata, errs := readAllFileInfo(s.ctx, disks, bucket, minioMetaMultipartBucket, candidate, "", false, false) if s.preflight { - // Readiness is stronger than listing liveness: even one readable legacy - // copy must be drained, and an unreadable copy cannot certify readiness. + // The diagnostic counts even a single readable legacy copy. It does not + // gate listing; unreadable copies still prevent a complete diagnosis. var oldest MultipartInfo legacy := false _, nativeID := multipartUploadTime(uploadUUID) @@ -176,10 +183,11 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket, } fi := metadata[i] storedBucket, storedObject, valid := er.multipartIdentity(fi, shaDir) - if storedBucket != "" && storedObject != "" && !valid { + isLegacy := multipartLegacyIdentity(fi) + if !isLegacy && (!valid || !nativeID) { return MultipartInfo{}, false, false, errMultipartListingIdentity } - if !valid || !nativeID { + if isLegacy { info := multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime) if !legacy || info.Initiated.Before(oldest.Initiated) { oldest = info @@ -203,16 +211,29 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket, if err != nil { return MultipartInfo{}, false, false, err } + // The generic metadata quorum does not compare these identity fields. + // Contradictory readable copies cannot certify a historical upload: picking + // a copy with missing fields must not silently hide a new-format record. + legacy := multipartLegacyIdentity(fi) storedBucket, storedObject, valid := er.multipartIdentity(fi, shaDir) + for i, other := range metadata { + if errs[i] != nil || !other.IsValid() { + continue + } + if multipartLegacyIdentity(other) != legacy || + (!legacy && (other.Metadata[multipartMetaBucket] != storedBucket || other.Metadata[multipartMetaObject] != storedObject)) { + return MultipartInfo{}, false, false, fmt.Errorf("%w: inconsistent copies at %s", errMultipartListingIdentity, candidate) + } + } info := multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime) - if storedBucket == "" || storedObject == "" { + if legacy { return info, false, true, nil } if !valid { return MultipartInfo{}, false, false, fmt.Errorf("%w: %s", errMultipartListingIdentity, candidate) } if _, ok := multipartUploadTime(uploadUUID); !ok { - return info, false, true, nil + return MultipartInfo{}, false, false, fmt.Errorf("%w: %s", errMultipartListingIdentity, candidate) } if bucket != "" && bucket != storedBucket { return MultipartInfo{}, false, false, nil @@ -220,7 +241,7 @@ func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket, return info, true, false, nil } -func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, poolIdx, setIdx int) ([]MultipartInfo, bool, error) { +func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, poolIdx, setIdx int) ([]MultipartInfo, error) { disks := er.getDisks() report := multipartScanSet{Pool: poolIdx, Set: setIdx, Drives: er.setDriveCount} var resultErr error @@ -272,7 +293,7 @@ func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, p resultErr = toObjectErr(errErasureReadQuorum, bucket) } if resultErr != nil { - return nil, false, resultErr + return nil, resultErr } type candidate struct { path string @@ -325,25 +346,23 @@ dispatch: if s.ctx.Err() != nil { resultErr = SlowDown{} } - return uploads, report.LegacyUploads != 0, resultErr + return uploads, resultErr } // multipartPreflight scans all pools, sets and drives, independent of caches. -// A majority suffices for normal listing; upgrade readiness requires every -// drive to have been inspected. The operator must also upgrade all writers. +// A majority suffices for normal listing; a complete diagnosis requires every +// drive to have been inspected. Ready describes the absence of legacy copies, +// not whether listing is enabled. func (z *erasureServerPools) multipartPreflight(ctx context.Context) (multipartPreflightReport, error) { s, err := startMultipartScan(ctx, true) if err != nil { return multipartPreflightReport{}, err } defer s.close() - report := multipartPreflightReport{Complete: true, Mode: "strict"} - if globalAPIConfig.getMultipartListingLegacy() { - report.Mode = "legacy" - } + report := multipartPreflightReport{Complete: true, Mode: "persistent"} for p, pool := range z.serverPools { for i, set := range pool.sets { - _, _, _ = set.scanMultipartUploads(s, "", p, i) + _, _ = set.scanMultipartUploads(s, "", p, i) } } report.Sets = s.sets diff --git a/cmd/erasure-multipart-listing_test.go b/cmd/erasure-multipart-listing_test.go index c669a6123..c481cbd67 100644 --- a/cmd/erasure-multipart-listing_test.go +++ b/cmd/erasure-multipart-listing_test.go @@ -31,7 +31,7 @@ func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) { if err != nil { t.Fatal(err) } - setMultipartListingTestMode(t, false) + var firstID, secondID string for attempt := 0; attempt < 32; attempt++ { one, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) @@ -148,28 +148,25 @@ func TestMultipartListingLegacyPreflight(t *testing.T) { t.Fatal(err) } z.mpCache.Clear() - t.Run("legacy-upgrade", func(t *testing.T) { - setMultipartListingTestMode(t, true) - exact, err := z.ListMultipartUploads(t.Context(), other, "old", "", "", "", 10) + for _, target := range []string{bucket, other} { + got, err := z.ListMultipartUploads(t.Context(), target, "", "", "", "", 10) if err != nil { t.Fatal(err) } - requireMultipartUploadKeys(t, exact, "old") - const empty = "multipart-legacy-empty" - if err := z.MakeBucket(t.Context(), empty, MakeBucketOptions{}); err != nil { - t.Fatal(err) + if target == bucket { + requireMultipartUploadKeys(t, got, "a") + } else { + requireMultipartUploadKeys(t, got) } - got, err := z.ListMultipartUploads(t.Context(), empty, "", "", "", "", 10) - if err != nil || len(got.Uploads) != 0 { - t.Fatalf("old upload in another bucket broke legacy listing: %+v %v", got, err) - } - }) - _, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10) - if !errors.Is(err, errMultipartListingLegacy) { - t.Fatalf("old upload in another bucket: %v", err) } + // Exact-looking prefixes now have the same durable semantics. + exact, err := z.ListMultipartUploads(t.Context(), other, "old", "", "", "", 10) + if err != nil { + t.Fatal(err) + } + requireMultipartUploadKeys(t, exact) report, err := z.multipartPreflight(t.Context()) - if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 { + if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" { t.Fatalf("legacy preflight: %+v %v", report, err) } if err = z.AbortMultipartUpload(t.Context(), other, "old", old.UploadID, ObjectOptions{}); err != nil { @@ -187,7 +184,7 @@ func TestMultipartListingLegacyPreflight(t *testing.T) { } func TestMultipartListingIdentityFallback(t *testing.T) { - for _, kind := range []string{"old", "corrupt", "read-failure", "wrong-bucket"} { + for _, kind := range []string{"old", "partial", "corrupt", "read-failure", "wrong-bucket"} { t.Run(kind, func(t *testing.T) { z, set, bucket := multipartListingFixture(t) if _, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil { @@ -209,8 +206,11 @@ func TestMultipartListingIdentityFallback(t *testing.T) { return FileInfo{}, errFileCorrupt } fi.Metadata = cloneMSS(fi.Metadata) - if kind == "old" { + if kind == "old" || kind == "partial" { delete(fi.Metadata, multipartMetaBucket) + if kind == "old" { + delete(fi.Metadata, multipartMetaObject) + } } else { fi.Metadata[multipartMetaBucket] = "another-bucket" } @@ -231,7 +231,7 @@ func TestMultipartListingIdentityFallback(t *testing.T) { func TestMultipartAbortPoolsAndRetry(t *testing.T) { z, bucket := consistencyPools(t) - setMultipartListingTestMode(t, false) + mp, err := z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) if err != nil { t.Fatal(err) @@ -252,10 +252,12 @@ func TestMultipartAbortPoolsAndRetry(t *testing.T) { if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 { t.Fatalf("unknown pool must not acknowledge cancellation: %v", err) } + if _, err := z.serverPools[1].GetMultipartInfo(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatalf("earlier pool error changed the later pool: %v", err) + } emptySet.getDisks = original - err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}) - if _, ok := err.(InvalidUploadID); !ok { - t.Fatalf("retry after all pools confirm absence: %v", err) + if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatalf("retry after pool recovery: %v", err) } mp, err = z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) if err != nil { @@ -289,12 +291,13 @@ func TestMultipartAbortRetryBelowReadQuorum(t *testing.T) { }} return disks } - if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err == nil { - t.Fatal("lost acknowledgement must fail") + if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatalf("best-effort cleanup must retain read-quorum availability: %v", err) } set.getDisks = original - if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { - t.Fatalf("one remaining metadata copy prevented retry: %v", err) + var invalid InvalidUploadID + if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) { + t.Fatalf("minority remnant must retain absent-upload result: %v", err) } } @@ -304,7 +307,7 @@ func TestMultipartListingMarkerHTTP(t *testing.T) { if err != nil { t.Fatal(err) } - setMultipartListingTestMode(t, false) + for _, tc := range []struct { key, marker string status int @@ -388,7 +391,7 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) { t.Fatal(err) } z := obj.(*erasureServerPools) - setMultipartListingTestMode(t, false) + t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) }) saved := globalStorageClass globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 8}}) @@ -417,9 +420,7 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) { } set.getDisks = func() []StorageAPI { disks := append([]StorageAPI(nil), original()...) - for i := 2; i < 9; i++ { - disks[i] = nil - } + disks[8] = nil return disks } if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { @@ -431,7 +432,7 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) { } } set.getDisks = original - _, metadata, err := set.checkUploadIDExists(t.Context(), bucket, "a", mp.UploadID, true) + _, metadata, err := set.checkUploadIDExists(t.Context(), bucket, "a", mp.UploadID, false) if err != nil { t.Fatalf("late creation boundary changed; review and remove the documented limitation: %v", err) } @@ -441,10 +442,10 @@ func TestMultipartAbortLateCreateBoundary(t *testing.T) { count++ } } - if count != 14 { - t.Fatalf("expected fourteen resurrected copies, got %d", count) + if count != 8 { + t.Fatalf("expected eight resurrected copies, got %d", count) } - t.Log("KNOWN UNRESOLVED BOUNDARY: seven delayed creation writes plus seven offline copies restore a writable upload after acknowledged cancellation") + t.Log("KNOWN UNRESOLVED BOUNDARY: seven delayed creation writes plus one offline copy restore a readable upload after acknowledged cancellation") } type multipartListingCountingDisk struct { @@ -474,7 +475,7 @@ func TestMultipartListingScanCosts(t *testing.T) { t.Fatal(err) } z := obj.(*erasureServerPools) - setMultipartListingTestMode(t, false) + t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) }) const bucket, otherBucket = "r9-scan-target", "r9-scan-unrelated" for _, name := range []string{bucket, otherBucket} { @@ -544,7 +545,7 @@ func multipartListingFixture(t *testing.T) (*erasureServerPools, *erasureObjects if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil { t.Fatal(err) } - setMultipartListingTestMode(t, false) + return z, z.serverPools[0].getHashedSet("a"), bucket } @@ -588,45 +589,6 @@ func TestMultipartListingMissingMarkers(t *testing.T) { } } -func TestMultipartListingAbortRecovery(t *testing.T) { - for _, offline := range []int{1, 2} { - t.Run(fmt.Sprint(offline), func(t *testing.T) { - z, set, bucket := multipartListingFixture(t) - mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) - if err != nil { - t.Fatal(err) - } - original := set.getDisks - t.Cleanup(func() { set.getDisks = original }) - set.getDisks = func() []StorageAPI { - disks := append([]StorageAPI(nil), original()...) - for i := 0; i < offline; i++ { - disks[i] = nil - } - return disks - } - err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}) - if offline == 1 && err != nil { - t.Fatal(err) - } - if offline == 2 && (err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503) { - t.Fatalf("two offline drives must not acknowledge cancellation: %v", err) - } - set.getDisks = original - if offline == 2 { - // Retry a partially completed deletion after the original disks return. - if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { - t.Fatal(err) - } - } - got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100) - if err != nil || len(got.Uploads) != 0 { - t.Fatalf("canceled upload reappeared: %+v %v", got, err) - } - }) - } -} - func TestMultipartListingCoverage(t *testing.T) { z, set, bucket := multipartListingFixture(t) mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) @@ -670,7 +632,7 @@ func TestMultipartListingBudgetAndAdmission(t *testing.T) { } defer scan.close() scan.remaining.Store(1) - _, _, err = set.scanMultipartUploads(scan, bucket, 0, 0) + _, err = set.scanMultipartUploads(scan, bucket, 0, 0) var limited SlowDown if !errors.As(err, &limited) { t.Fatalf("budget: %v", err) @@ -695,7 +657,7 @@ func TestMultipartListingAdmissionHTTP(t *testing.T) { if err != nil { t.Fatal(err) } - setMultipartListingTestMode(t, false) + for range cap(multipartScanSlots) { scan, err := startMultipartScan(t.Context(), false) if err != nil { @@ -743,7 +705,7 @@ func TestMultipartListingPreflightMinorityLegacy(t *testing.T) { } } report, err := z.multipartPreflight(t.Context()) - if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 { + if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" { t.Fatalf("minority legacy copy must prevent readiness: %+v %v", report, err) } } diff --git a/cmd/erasure-multipart-migration_test.go b/cmd/erasure-multipart-migration_test.go index db74fe432..cc642aef2 100644 --- a/cmd/erasure-multipart-migration_test.go +++ b/cmd/erasure-multipart-migration_test.go @@ -12,41 +12,13 @@ import ( "sync/atomic" "testing" - "github.com/minio/minio/internal/config/api" "github.com/minio/minio/internal/config/storageclass" "github.com/minio/minio/internal/dsync" "github.com/minio/minio/internal/grid" xnet "github.com/pgsty/silo-pkg/v3/net" ) -func setMultipartListingTestMode(t *testing.T, legacy bool) { - t.Helper() - globalAPIConfig.mu.Lock() - previous := globalAPIConfig.multipartListingStrict - globalAPIConfig.multipartListingStrict = !legacy - globalAPIConfig.mu.Unlock() - t.Cleanup(func() { - globalAPIConfig.mu.Lock() - globalAPIConfig.multipartListingStrict = previous - globalAPIConfig.mu.Unlock() - }) -} - -func TestMultipartListingDefaultMode(t *testing.T) { - var uninitialized apiConfig - if !uninitialized.getMultipartListingLegacy() { - t.Fatal("runtime zero value enabled strict mode before config initialization") - } - for _, mode := range []string{"", "legacy", "strict"} { - var local apiConfig - local.init(api.Config{RequestsMax: 1, MultipartListing: mode}, []int{4}, false) - if local.getMultipartListingLegacy() != (mode != "strict") { - t.Fatalf("unexpected mode for %q", mode) - } - } -} - -func TestMultipartAbortLegacyAvailability(t *testing.T) { +func TestMultipartAbortDefaultAvailability(t *testing.T) { for _, drives := range []int{4, 16} { for _, offline := range []int{0, 1, drives / 2} { t.Run(fmt.Sprintf("drives=%d/offline=%d", drives, offline), func(t *testing.T) { @@ -56,7 +28,7 @@ func TestMultipartAbortLegacyAvailability(t *testing.T) { } z := obj.(*erasureServerPools) t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) }) - setMultipartListingTestMode(t, true) + previous := globalStorageClass globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: drives / 2}}) t.Cleanup(func() { globalStorageClass.Update(previous) }) @@ -91,11 +63,11 @@ func TestMultipartAbortLegacyAvailability(t *testing.T) { } } -func TestMultipartAbortLegacyPoolOrder(t *testing.T) { +func TestMultipartAbortDefaultPoolOrder(t *testing.T) { for _, owner := range []int{0, 1} { t.Run(fmt.Sprintf("owner=%d", owner), func(t *testing.T) { z, bucket := consistencyPools(t) - setMultipartListingTestMode(t, true) + mp, err := z.serverPools[owner].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) if err != nil { t.Fatal(err) @@ -106,7 +78,7 @@ func TestMultipartAbortLegacyPoolOrder(t *testing.T) { emptySet.getDisks = func() []StorageAPI { return make([]StorageAPI, emptySet.setDriveCount) } err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}) if owner == 0 && err != nil { - t.Fatalf("unrelated later pool blocked legacy cancellation: %v", err) + t.Fatalf("unrelated later pool blocked cancellation: %v", err) } if owner == 1 { if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 { @@ -124,123 +96,56 @@ func TestMultipartAbortLegacyPoolOrder(t *testing.T) { } } -func TestMultipartAbortMinorityLegacyCleanup(t *testing.T) { - for _, failDelete := range []bool{false, true} { - t.Run(fmt.Sprintf("delete-fails=%v", failDelete), func(t *testing.T) { - z, set, bucket := multipartListingFixture(t) - mp, err := z.NewMultipartUpload(t.Context(), bucket, "old", ObjectOptions{}) - if err != nil { - t.Fatal(err) - } - path := set.getUploadIDDir(bucket, "old", mp.UploadID) - original := set.getDisks - disks := original() - fi, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, path, "", ReadOptions{}) - if err != nil { - t.Fatal(err) - } - delete(fi.Metadata, multipartMetaBucket) - delete(fi.Metadata, multipartMetaObject) - if err := disks[0].WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, path, fi); err != nil { - t.Fatal(err) - } - for _, d := range disks[1:] { - if err := d.Delete(t.Context(), minioMetaMultipartBucket, path, DeleteOptions{Recursive: true}); err != nil { - t.Fatal(err) - } - } - report, err := z.multipartPreflight(t.Context()) - if err != nil || report.Ready || report.LegacyUploads != 1 { - t.Fatalf("invalid legacy fixture: %+v %v", report, err) - } - if failDelete { - set.getDisks = func() []StorageAPI { - wrapped := append([]StorageAPI(nil), disks...) - wrapped[0] = multipartListingFaultDisk{StorageAPI: disks[0], delete: func(context.Context, string, string, DeleteOptions) error { return errFaultyDisk }} - return wrapped - } - t.Cleanup(func() { set.getDisks = original }) - err := z.AbortMultipartUpload(t.Context(), bucket, "old", mp.UploadID, ObjectOptions{}) - if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 { - t.Fatalf("empty drives masked failed deletion of the observed replica: %v", err) - } - if _, present := z.mpCache.Load(mp.UploadID); !present { - t.Fatal("failed cancellation evicted upload from cache") - } - set.getDisks = original - } - if err := z.AbortMultipartUpload(t.Context(), bucket, "old", mp.UploadID, ObjectOptions{}); err != nil { - t.Fatal(err) - } - report, err = z.multipartPreflight(t.Context()) - if err != nil || !report.Ready || report.LegacyUploads != 0 { - t.Fatalf("acknowledged cleanup left online legacy remnants: %+v %v", report, err) - } - if _, present := z.mpCache.Load(mp.UploadID); present { - t.Fatal("successful cancellation retained cache entry") - } - }) - } -} - func TestMultipartAbortWrongTargetKeepsCache(t *testing.T) { - for _, legacy := range []bool{true, false} { - t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) { - z, _, bucket := multipartListingFixture(t) - setMultipartListingTestMode(t, legacy) - const other = "multipart-other-bucket" - if err := z.MakeBucket(t.Context(), other, MakeBucketOptions{}); err != nil { - t.Fatal(err) - } - mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid-key", ObjectOptions{}) - if err != nil { - t.Fatal(err) - } - for _, target := range [][2]string{{bucket, "wrong-key"}, {other, "valid-key"}} { - err := z.AbortMultipartUpload(t.Context(), target[0], target[1], mp.UploadID, ObjectOptions{}) - var invalid InvalidUploadID - if !errors.As(err, &invalid) { - t.Fatalf("wrong target result: %v", err) - } - if _, present := z.mpCache.Load(mp.UploadID); !present { - t.Fatal("wrong-target abort evicted another upload") - } - if _, err := z.GetMultipartInfo(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); err != nil { - t.Fatalf("wrong-target abort harmed valid upload: %v", err) - } - } - if err := z.AbortMultipartUpload(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); err != nil { - t.Fatal(err) - } - var invalid InvalidUploadID - if err := z.AbortMultipartUpload(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) { - t.Fatalf("repeat abort: %v", err) - } - }) + z, _, bucket := multipartListingFixture(t) + + const other = "multipart-other-bucket" + if err := z.MakeBucket(t.Context(), other, MakeBucketOptions{}); err != nil { + t.Fatal(err) + } + mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid-key", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + for _, target := range [][2]string{{bucket, "wrong-key"}, {other, "valid-key"}} { + err := z.AbortMultipartUpload(t.Context(), target[0], target[1], mp.UploadID, ObjectOptions{}) + var invalid InvalidUploadID + if !errors.As(err, &invalid) { + t.Fatalf("wrong target result: %v", err) + } + if _, present := z.mpCache.Load(mp.UploadID); !present { + t.Fatal("wrong-target abort evicted another upload") + } + if _, err := z.GetMultipartInfo(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatalf("wrong-target abort harmed valid upload: %v", err) + } + } + if err := z.AbortMultipartUpload(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatal(err) + } + var invalid InvalidUploadID + if err := z.AbortMultipartUpload(t.Context(), bucket, "valid-key", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) { + t.Fatalf("repeat abort: %v", err) } } func TestMultipartAbortRejectsUnsafeID(t *testing.T) { - for _, legacy := range []bool{true, false} { - t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) { - z, _, bucket := multipartListingFixture(t) - setMultipartListingTestMode(t, legacy) - mp, err := z.NewMultipartUpload(t.Context(), bucket, "keep", ObjectOptions{}) - if err != nil { - t.Fatal(err) - } - for _, suffix := range []string{".", "..", "../other", "a/b", "a\\b", ""} { - id := base64.RawURLEncoding.EncodeToString([]byte("deployment." + suffix)) - err := z.AbortMultipartUpload(t.Context(), bucket, "keep", id, ObjectOptions{}) - var invalid InvalidUploadID - if !errors.As(err, &invalid) { - t.Fatalf("unsafe ID accepted: suffix=%q err=%v", suffix, err) - } - if _, err := z.GetMultipartInfo(t.Context(), bucket, "keep", mp.UploadID, ObjectOptions{}); err != nil { - t.Fatalf("valid upload harmed by rejected ID: %v", err) - } - } - }) + z, _, bucket := multipartListingFixture(t) + + mp, err := z.NewMultipartUpload(t.Context(), bucket, "keep", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + for _, suffix := range []string{".", "..", "../other", "a/b", "a\\b", ""} { + id := base64.RawURLEncoding.EncodeToString([]byte("deployment." + suffix)) + err := z.AbortMultipartUpload(t.Context(), bucket, "keep", id, ObjectOptions{}) + var invalid InvalidUploadID + if !errors.As(err, &invalid) { + t.Fatalf("unsafe ID accepted: suffix=%q err=%v", suffix, err) + } + if _, err := z.GetMultipartInfo(t.Context(), bucket, "keep", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatalf("valid upload harmed by rejected ID: %v", err) + } } } @@ -275,34 +180,30 @@ func TestMultipartAbortPeerNotification(t *testing.T) { set.nsMutex = newNSLock(true) set.getLockers = func() ([]dsync.NetLocker, string) { return lockers, "multipart-test" } t.Cleanup(func() { set.nsMutex, set.getLockers = oldMutex, oldLockers }) - for _, legacy := range []bool{true, false} { - t.Run(fmt.Sprintf("legacy=%v", legacy), func(t *testing.T) { - setMultipartListingTestMode(t, legacy) - for range 4 { - mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid", ObjectOptions{}) - if err != nil { - t.Fatal(err) - } - before := notifications.Load() - var invalid InvalidUploadID - if err := z.AbortMultipartUpload(t.Context(), bucket, "wrong", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) { - t.Fatalf("wrong target: %v", err) - } - if notifications.Load() != before { - t.Fatal("wrong-target abort sent a destructive peer notification") - } - if err := z.AbortMultipartUpload(t.Context(), bucket, "valid", mp.UploadID, ObjectOptions{}); err != nil { - t.Fatal(err) - } - if notifications.Load() != before+1 { - t.Fatal("successful abort lost peer notification after unlocking") - } - } - }) + + for range 4 { + mp, err := z.NewMultipartUpload(t.Context(), bucket, "valid", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + before := notifications.Load() + var invalid InvalidUploadID + if err := z.AbortMultipartUpload(t.Context(), bucket, "wrong", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) { + t.Fatalf("wrong target: %v", err) + } + if notifications.Load() != before { + t.Fatal("wrong-target abort sent a destructive peer notification") + } + if err := z.AbortMultipartUpload(t.Context(), bucket, "valid", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatal(err) + } + if notifications.Load() != before+1 { + t.Fatal("successful abort lost peer notification after unlocking") + } } } -func TestMultipartAbortStrictMajority(t *testing.T) { +func TestMultipartAbortBestEffortCleanup(t *testing.T) { z, set, bucket := multipartListingFixture(t) mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) if err != nil { @@ -317,15 +218,19 @@ func TestMultipartAbortStrictMajority(t *testing.T) { } t.Cleanup(func() { set.getDisks = original }) if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { - t.Fatalf("ordinary majority cancellation was tightened: %v", err) + t.Fatalf("read-quorum cancellation was tightened: %v", err) } - // One failed deletion is tolerated in the ordinary majority path. Retrying - // after recovery must also clean the now-minority remnant. + // Best-effort deletion may leave a remnant. Retrying after recovery cleans + // it, but the read quorum no longer recognizes it as an active upload. if _, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, set.getUploadIDDir(bucket, "a", mp.UploadID), "", ReadOptions{}); err != nil { t.Fatalf("fault fixture did not leave a remnant: %v", err) } set.getDisks = original - if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { - t.Fatal(err) + var invalid InvalidUploadID + if err := z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); !errors.As(err, &invalid) { + t.Fatalf("minority retry must preserve default absent-upload result: %v", err) + } + if _, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, set.getUploadIDDir(bucket, "a", mp.UploadID), "", ReadOptions{}); !errors.Is(err, errFileNotFound) { + t.Fatalf("best-effort retry did not clean remnant: %v", err) } } diff --git a/cmd/erasure-multipart-unified_test.go b/cmd/erasure-multipart-unified_test.go new file mode 100644 index 000000000..a14a0ea73 --- /dev/null +++ b/cmd/erasure-multipart-unified_test.go @@ -0,0 +1,256 @@ +// Copyright (c) 2026 Ruohang Feng +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "context" + "encoding/xml" + "errors" + "net/http" + "net/http/httptest" + "slices" + "testing" +) + +func rewriteMultipartIdentity(t testing.TB, set *erasureObjects, bucket, key, id string, change func(map[string]string)) { + t.Helper() + fi, metadata, err := set.checkUploadIDExists(t.Context(), bucket, key, id, true) + if err != nil { + t.Fatal(err) + } + for i := range metadata { + change(metadata[i].Metadata) + } + if _, err := writeAllMetadata(t.Context(), set.getDisks(), bucket, minioMetaMultipartBucket, + set.getUploadIDDir(bucket, key, id), metadata, fi.WriteQuorum(set.defaultWQuorum())); err != nil { + t.Fatal(err) + } +} + +func removeMultipartIdentity(metadata map[string]string) { + delete(metadata, multipartMetaBucket) + delete(metadata, multipartMetaObject) +} + +// The actual S3 handler must use the same semantics for every prefix and page, +// even with old records and a stale, misleading node-local cache. +func TestMultipartListingMixedHTTP(t *testing.T) { + z, set, _ := multipartListingFixture(t) + bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{}) + if err != nil { + t.Fatal(err) + } + for _, key := range []string{"dir/old", "dir/a", "dir/sub/b", "outside"} { + mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + if key == "dir/old" { + rewriteMultipartIdentity(t, set, bucket, key, mp.UploadID, removeMultipartIdentity) + } + } + z.mpCache.Clear() + z.mpCache.Store("stale", MultipartInfo{Bucket: bucket, Object: "ghost", UploadID: "stale"}) + list := func(prefix, keyMarker, uploadMarker, delimiter string) ListMultipartUploadsResponse { + t.Helper() + req, err := newTestSignedRequestV4(http.MethodGet, + getListMultipartUploadsURLWithParams("", bucket, prefix, keyMarker, uploadMarker, delimiter, "1"), + 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + var got ListMultipartUploadsResponse + if rec.Code != http.StatusOK { + t.Fatalf("list %q: HTTP %d %s", prefix, rec.Code, rec.Body.String()) + } + if err := xml.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + return got + } + var keys []string + keyMarker, uploadMarker := "", "" + for page := 0; ; page++ { + if page >= 4 { + t.Fatal("pagination did not terminate") + } + got := list("", keyMarker, uploadMarker, "") + if len(got.Uploads) != 1 { + t.Fatalf("page size: %+v", got) + } + keys = append(keys, got.Uploads[0].Key) + if !got.IsTruncated { + break + } + keyMarker, uploadMarker = got.NextKeyMarker, got.NextUploadIDMarker + } + if !slices.Equal(keys, []string{"dir/a", "dir/sub/b", "outside"}) { + t.Fatalf("mixed full listing: %v", keys) + } + first := list("dir/", "", "", "/") + if len(first.Uploads) != 1 || first.Uploads[0].Key != "dir/a" || !first.IsTruncated { + t.Fatalf("prefix first page: %+v", first) + } + rest := list("dir/", first.NextKeyMarker, first.NextUploadIDMarker, "/") + if len(rest.Uploads) != 0 || rest.IsTruncated || len(rest.CommonPrefixes) != 1 || rest.CommonPrefixes[0].Prefix != "dir/sub/" { + t.Fatalf("prefix continuation: %+v", rest) + } + if got := list("dir/old", "", "", ""); len(got.Uploads) != 0 || got.IsTruncated { + t.Fatalf("historical exact-looking prefix: %+v", got) + } +} + +func TestMultipartListingInvalidIdentity(t *testing.T) { + for _, kind := range []string{"missing-bucket", "missing-object", "empty-bucket", "empty-object", "empty-both", "wrong-hash", "read-failure"} { + t.Run(kind, func(t *testing.T) { + z, set, bucket := multipartListingFixture(t) + mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + rewriteMultipartIdentity(t, set, bucket, "a", mp.UploadID, func(metadata map[string]string) { + switch kind { + case "missing-bucket": + delete(metadata, multipartMetaBucket) + case "missing-object": + delete(metadata, multipartMetaObject) + case "empty-bucket": + metadata[multipartMetaBucket] = "" + case "empty-object": + metadata[multipartMetaObject] = "" + case "empty-both": + metadata[multipartMetaBucket], metadata[multipartMetaObject] = "", "" + case "wrong-hash": + metadata[multipartMetaObject] = "wrong-key" + case "read-failure": + removeMultipartIdentity(metadata) + } + }) + if kind == "read-failure" { + original := set.getDisks + t.Cleanup(func() { set.getDisks = original }) + set.getDisks = func() []StorageAPI { + disks := append([]StorageAPI(nil), original()...) + for i, disk := range disks { + disks[i] = multipartListingFaultDisk{StorageAPI: disk, read: func(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) { + return FileInfo{}, errFaultyDisk + }} + } + return disks + } + } + got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10) + if err == nil || len(got.Uploads) != 0 || toAPIError(t.Context(), err).HTTPStatusCode != http.StatusServiceUnavailable { + t.Fatalf("invalid/unreadable record returned a successful page: %+v %v", got, err) + } + if kind != "read-failure" && !errors.Is(err, errMultipartListingIdentity) { + t.Fatalf("identity error: %v", err) + } + report, err := z.multipartPreflight(t.Context()) + if err != nil || report.Complete || report.Ready || report.LegacyUploads != 0 { + t.Fatalf("invalid/unreadable record counted as historical: %+v %v", report, err) + } + }) + } +} + +func TestMultipartListingRestart(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + obj, dirs, err := prepareErasure(ctx, 4) + if err != nil { + cancel() + t.Fatal(err) + } + t.Cleanup(func() { cancel(); obj.Shutdown(context.Background()); removeRoots(dirs) }) + const bucket, key = "multipart-restart", "dir/new" + if err := obj.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil { + t.Fatal(err) + } + mp, err := obj.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + cancel() + if err := obj.Shutdown(context.Background()); err != nil { + t.Fatal(err) + } + reopened, _, err := initObjectLayer(t.Context(), mustGetPoolEndpoints(0, dirs...)) + if err != nil { + t.Fatal(err) + } + obj = reopened + if _, cached := obj.(*erasureServerPools).mpCache.Load(mp.UploadID); cached { + t.Fatal("restart unexpectedly preserved the upload cache") + } + got, err := obj.ListMultipartUploads(t.Context(), bucket, "dir/", "", "", "", 10) + if err != nil { + t.Fatal(err) + } + requireMultipartUploadKeys(t, got, key) + if got.Uploads[0].UploadID != mp.UploadID { + t.Fatal("restart changed upload identity") + } +} + +func TestMultipartListingConflictingIdentityCopies(t *testing.T) { + for _, oldCopies := range []int{1, 3} { + z, set, bucket := multipartListingFixture(t) + mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + path := set.getUploadIDDir(bucket, "a", mp.UploadID) + for _, disk := range set.getDisks()[:oldCopies] { + fi, err := disk.ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, path, "", ReadOptions{}) + if err != nil { + t.Fatal(err) + } + removeMultipartIdentity(fi.Metadata) + if err := disk.WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, path, fi); err != nil { + t.Fatal(err) + } + } + got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10) + if !errors.Is(err, errMultipartListingIdentity) || len(got.Uploads) != 0 { + t.Fatalf("%d contradictory copies returned a successful page: %+v %v", oldCopies, got, err) + } + } +} + +func TestMultipartListingUnsupportedStoredID(t *testing.T) { + for _, historical := range []bool{false, true} { + z, set, bucket := multipartListingFixture(t) + mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + oldPath := set.getUploadIDDir(bucket, "a", mp.UploadID) + unsupportedPath := pathJoin(set.getMultipartSHADir(bucket, "a"), "unsupported-id") + for _, disk := range set.getDisks() { + fi, err := disk.ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, oldPath, "", ReadOptions{}) + if err != nil { + t.Fatal(err) + } + if historical { + removeMultipartIdentity(fi.Metadata) + } + if err := disk.WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, unsupportedPath, fi); err != nil { + t.Fatal(err) + } + if err := disk.Delete(t.Context(), minioMetaMultipartBucket, oldPath, DeleteOptions{Recursive: true}); err != nil { + t.Fatal(err) + } + } + got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10) + if historical { + if err != nil || len(got.Uploads) != 0 { + t.Fatalf("unidentified historical ID: %+v %v", got, err) + } + } else if !errors.Is(err, errMultipartListingIdentity) { + t.Fatalf("invalid new ID silently treated as historical: %+v %v", got, err) + } + } +} diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go index 5e57cbd17..828d3619b 100644 --- a/cmd/erasure-multipart.go +++ b/cmd/erasure-multipart.go @@ -414,7 +414,7 @@ func paginateMultipartUploads(uploads []MultipartInfo, prefix, keyMarker, upload } // listMultipartUploadsExact preserves the hashed exact-object lookup used by -// multipart write placement and by rolling-upgrade legacy mode. +// multipart write placement. Public listing always uses the durable scan. func (er erasureObjects) listMultipartUploadsExact(ctx context.Context, bucket, object, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) { auditObjectErasureSet(ctx, "ListMultipartUploads", object, &er) @@ -527,13 +527,10 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, prefi return ListMultipartsInfo{}, err } defer scan.close() - uploads, legacy, err := er.scanMultipartUploads(scan, bucket, 0, 0) + uploads, err := er.scanMultipartUploads(scan, bucket, 0, 0) if err != nil { return ListMultipartsInfo{}, err } - if legacy { - return ListMultipartsInfo{}, errMultipartListingLegacy - } return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil } @@ -1788,11 +1785,9 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str return fi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended), nil } -// abortMultipartUpload retains read-quorum validation and best-effort cleanup -// in legacy mode. Strict mode requires majority deletion acknowledgements and -// permits retrying remnants below read quorum. Neither mode fences creation -// writes still executing after a storage timeout. -func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions, legacy bool) (bool, error) { +// abortMultipartUpload retains read-quorum validation and best-effort cleanup. +// It does not fence creation writes still executing after a storage timeout. +func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (bool, error) { if !opts.NoAuditLog { auditObjectErasureSet(ctx, "AbortMultipartUpload", object, &er) } @@ -1804,70 +1799,19 @@ func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, objec if !ok || internalID == "" || internalID == "." || internalID == ".." || strings.ContainsAny(internalID, "/\\") { return false, InvalidUploadID{Bucket: bucket, Object: object, UploadID: uploadID} } - if legacy { - // Keep the released read-quorum and best-effort cleanup behavior. - // The upload ID safety check above applies to both modes. - defer er.deleteAll(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID)) - _, _, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, false) - err = toObjectErr(err, bucket, object, uploadID) - if _, absent := err.(InvalidUploadID); absent { - return false, nil - } - return err == nil, err - } - disks := er.getDisks() - uploadPath := er.getUploadIDDir(bucket, object, uploadID) - _, errs := readAllFileInfo(ctx, disks, bucket, minioMetaMultipartBucket, uploadPath, "", false, false) - quorum := er.setDriveCount/2 + 1 - found, absent := false, 0 - observed := make([]bool, len(disks)) - for i, err := range errs { - switch { - case err == nil, errors.Is(err, errFileCorrupt): - found = true - observed[i] = true - case errors.Is(err, errFileNotFound), errors.Is(err, errFileVersionNotFound): - absent++ - } - } - if absent >= quorum && !found { + defer er.deleteAll(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID)) + _, _, err = er.checkUploadIDExists(ctx, bucket, object, uploadID, false) + err = toObjectErr(err, bucket, object, uploadID) + if _, absent := err.(InvalidUploadID); absent { return false, nil } - if !found { - return false, toObjectErr(errErasureReadQuorum, bucket, object, uploadID) - } - g := errgroup.WithNErrs(len(disks)) - for i, disk := range disks { - g.Go(func() error { - if disk == nil { - return errDiskNotFound - } - err := disk.Delete(ctx, minioMetaMultipartBucket, uploadPath, DeleteOptions{Recursive: true, Immediate: false}) - if errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound) { - return nil - } - return err - }, i) - } - deleteErrs := g.Wait() - if err := reduceWriteQuorumErrs(ctx, deleteErrs, nil, quorum); err != nil { - return true, toObjectErr(err, bucket, object, uploadID) - } - if absent >= quorum { - // Missing disks must not mask a failed cleanup of known remnants. - for i, found := range observed { - if found && deleteErrs[i] != nil { - return true, toObjectErr(errErasureWriteQuorum, bucket, object, uploadID) - } - } - } - return true, nil + return err == nil, err } -// AbortMultipartUpload cancels an upload using the configured mode. Offline -// part data may still need stale-upload cleanup after its drives return. +// AbortMultipartUpload cancels an upload using read-quorum validation. +// Offline part data may still need stale-upload cleanup after its drives return. func (er erasureObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (err error) { - found, err := er.abortMultipartUpload(ctx, bucket, object, uploadID, opts, globalAPIConfig.getMultipartListingLegacy()) + found, err := er.abortMultipartUpload(ctx, bucket, object, uploadID, opts) if err != nil { return err } diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 137204f97..e08b9be3c 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1916,9 +1916,6 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil { return ListMultipartsInfo{}, toObjectErr(err, bucket) } - if globalAPIConfig.getMultipartListingLegacy() { - return z.listMultipartUploadsLegacy(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads) - } scan, err := startMultipartScan(ctx, false) if err != nil { return ListMultipartsInfo{}, err @@ -1926,70 +1923,22 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p defer scan.close() var uploads []MultipartInfo - var keyless bool for idx, pool := range z.serverPools { if z.IsSuspended(idx) { continue } - poolUploads, poolKeyless, err := pool.scanMultipartUploads(scan, bucket, idx) + poolUploads, err := pool.scanMultipartUploads(scan, bucket, idx) if err != nil { return ListMultipartsInfo{}, err } uploads = append(uploads, poolUploads...) - keyless = keyless || poolKeyless } - // The old format cannot be enumerated authoritatively. Migration mode is - // explicit: another bucket must never silently change this API's semantics. - if keyless { - return ListMultipartsInfo{}, errMultipartListingLegacy - } + // Confirmed historical records lack a recoverable bucket/key and are + // omitted by the scan. They do not block listing identifiable uploads. return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil } -func (z *erasureServerPools) listMultipartUploadsLegacy(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (ListMultipartsInfo, error) { - poolResult := ListMultipartsInfo{} - poolResult.MaxUploads = maxUploads - poolResult.KeyMarker = keyMarker - poolResult.Prefix = prefix - poolResult.Delimiter = delimiter - - // if no prefix provided, return the list from cache - if prefix == "" { - if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil { - return ListMultipartsInfo{}, toObjectErr(err, bucket) - } - - z.mpCache.Range(func(_ string, mp MultipartInfo) bool { - if mp.Bucket == bucket { - poolResult.Uploads = append(poolResult.Uploads, mp) - } - return true - }) - sort.Slice(poolResult.Uploads, func(i int, j int) bool { - return poolResult.Uploads[i].Initiated.Before(poolResult.Uploads[j].Initiated) - }) - return poolResult, nil - } - - if z.SinglePool() { - return z.serverPools[0].getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads) - } - - for idx, pool := range z.serverPools { - if z.IsSuspended(idx) { - continue - } - result, err := pool.getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads) - if err != nil { - return result, err - } - poolResult.Uploads = append(poolResult.Uploads, result.Uploads...) - } - - return poolResult, nil -} - // Initiate a new multipart upload on a hashedSet based on object name. func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (mp *NewMultipartUploadResult, err error) { if err := checkNewMultipartArgs(ctx, bucket, object); err != nil { @@ -2199,28 +2148,15 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o ctx = lkctx.Context() defer lk.Unlock(lkctx) - legacy := globalAPIConfig.getMultipartListingLegacy() - found := false - var firstErr error for idx, pool := range z.serverPools { if z.IsSuspended(idx) { continue } - poolFound, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts, legacy) - if legacy && (poolFound || err != nil) { - // Match the released first-matching-pool behavior. + found, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts) + if found || err != nil { + // Preserve the default first-matching-pool behavior. return err } - found = found || poolFound - if err != nil && firstErr == nil { - firstErr = err - } - } - if firstErr != nil { - return firstErr - } - if found { - return nil } return InvalidUploadID{ Bucket: bucket, diff --git a/cmd/erasure-sets.go b/cmd/erasure-sets.go index d8b4dcbf6..4ee1a1f3c 100644 --- a/cmd/erasure-sets.go +++ b/cmd/erasure-sets.go @@ -894,28 +894,23 @@ func (s *erasureSets) ListMultipartUploads(ctx context.Context, bucket, prefix, return ListMultipartsInfo{}, err } defer scan.close() - uploads, legacy, err := s.scanMultipartUploads(scan, bucket, 0) + uploads, err := s.scanMultipartUploads(scan, bucket, 0) if err != nil { return ListMultipartsInfo{}, err } - if legacy { - return ListMultipartsInfo{}, errMultipartListingLegacy - } return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil } -func (s *erasureSets) scanMultipartUploads(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, bool, error) { +func (s *erasureSets) scanMultipartUploads(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, error) { var uploads []MultipartInfo - var keyless bool for i, set := range s.sets { - setUploads, setKeyless, err := set.scanMultipartUploads(scan, bucket, poolIdx, i) + setUploads, err := set.scanMultipartUploads(scan, bucket, poolIdx, i) if err != nil { - return nil, false, err + return nil, err } uploads = append(uploads, setUploads...) - keyless = keyless || setKeyless } - return uploads, keyless, nil + return uploads, nil } func (s *erasureSets) listMultipartUploadsExact(ctx context.Context, bucket, object string) (ListMultipartsInfo, error) { diff --git a/cmd/handler-api.go b/cmd/handler-api.go index 0b33bbcab..09790a60a 100644 --- a/cmd/handler-api.go +++ b/cmd/handler-api.go @@ -50,7 +50,6 @@ type apiConfig struct { transitionWorkers int staleUploadsExpiry time.Duration - multipartListingStrict bool staleUploadsCleanupInterval time.Duration deleteCleanupInterval time.Duration enableODirect bool @@ -182,7 +181,6 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int, legacy bool) { t.transitionWorkers = cfg.TransitionWorkers t.staleUploadsExpiry = cfg.StaleUploadsExpiry - t.multipartListingStrict = cfg.MultipartListing == "strict" t.deleteCleanupInterval = cfg.DeleteCleanupInterval t.enableODirect = cfg.EnableODirect t.gzipObjects = cfg.GzipObjects @@ -208,12 +206,6 @@ func (t *apiConfig) odirectEnabled() bool { return t.enableODirect } -func (t *apiConfig) getMultipartListingLegacy() bool { - t.mu.RLock() - defer t.mu.RUnlock() - return !t.multipartListingStrict -} - func (t *apiConfig) shouldGzipObjects() bool { t.mu.RLock() defer t.mu.RUnlock() diff --git a/cmd/list-multipart-uploads-compat_test.go b/cmd/list-multipart-uploads-compat_test.go index 4ff4c069f..0c8722c74 100644 --- a/cmd/list-multipart-uploads-compat_test.go +++ b/cmd/list-multipart-uploads-compat_test.go @@ -19,7 +19,6 @@ package cmd import ( "bytes" - "errors" "fmt" "slices" "testing" @@ -47,7 +46,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) { t.Fatal(err) } z := obj.(*erasureServerPools) - setMultipartListingTestMode(t, false) + t.Cleanup(func() { z.Shutdown(t.Context()) removeRoots(dirs) @@ -182,8 +181,8 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) { } // Simulate an upload written by a pre-upgrade server. Its key cannot be - // recovered by scanning the hashed namespace. Strict listing must fail; - // the old path is available only through an explicit migration setting. + // recovered by scanning the hashed namespace. Omit only that record, and + // preserve normal operations when its bucket, key and upload ID are known. legacyObject := objects[1] er := sets.getHashedSet(legacyObject) fi, metadata, err := er.checkUploadIDExists(t.Context(), bucket, legacyObject, uploadIDs[legacyObject], true) @@ -198,16 +197,25 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) { er.getUploadIDDir(bucket, legacyObject, uploadIDs[legacyObject]), metadata, fi.WriteQuorum(er.defaultWQuorum())); err != nil { t.Fatal(err) } - _, err = z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100) - if !errors.Is(err, errMultipartListingLegacy) { - t.Fatalf("legacy strict listing: %v", err) - } - setMultipartListingTestMode(t, true) legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100) if err != nil { t.Fatal(err) } - requireMultipartUploadKeys(t, legacy, legacyObject) + requireMultipartUploadKeys(t, legacy) + remaining, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100) + if err != nil { + t.Fatal(err) + } + requireMultipartUploadKeys(t, remaining, objects[2:]...) + part, err = z.PutObjectPart(t.Context(), bucket, legacyObject, uploadIDs[legacyObject], 1, + mustGetPutObjReader(t, bytes.NewBufferString("old part"), 8, "", ""), ObjectOptions{}) + if err != nil { + t.Fatalf("continue historical upload: %v", err) + } + if _, err = z.CompleteMultipartUpload(t.Context(), bucket, legacyObject, uploadIDs[legacyObject], + []CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{}); err != nil { + t.Fatalf("complete historical upload: %v", err) + } } func TestPaginateMultipartUploads(t *testing.T) { @@ -260,7 +268,7 @@ func TestPaginateMultipartUploads(t *testing.T) { func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) { z, bucket := consistencyPools(t) - setMultipartListingTestMode(t, false) + objects := []string{"a/one", "b/two", "c/three", "d/four"} for i, object := range objects { if _, err := z.serverPools[i%len(z.serverPools)].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{}); err != nil { diff --git a/internal/config/api/api.go b/internal/config/api/api.go index 0337f1008..5bd7cd8c1 100644 --- a/internal/config/api/api.go +++ b/internal/config/api/api.go @@ -68,7 +68,7 @@ const ( EnvAPIStaleUploadsCleanupInterval = "MINIO_API_STALE_UPLOADS_CLEANUP_INTERVAL" EnvAPIStaleUploadsExpiry = "MINIO_API_STALE_UPLOADS_EXPIRY" - EnvAPIMultipartListing = "MINIO_API_MULTIPART_LISTING" + EnvAPIMultipartListing = "MINIO_API_MULTIPART_LISTING" // Retired; used only for a startup warning. EnvAPIDeleteCleanupInterval = "MINIO_API_DELETE_CLEANUP_INTERVAL" EnvDeleteCleanupInterval = "MINIO_DELETE_CLEANUP_INTERVAL" EnvAPIODirect = "MINIO_API_ODIRECT" @@ -180,7 +180,6 @@ type Config struct { TransitionWorkers int `json:"transition_workers"` StaleUploadsCleanupInterval time.Duration `json:"stale_uploads_cleanup_interval"` StaleUploadsExpiry time.Duration `json:"stale_uploads_expiry"` - MultipartListing string `json:"multipart_listing"` DeleteCleanupInterval time.Duration `json:"delete_cleanup_interval"` EnableODirect bool `json:"enable_odirect"` GzipObjects bool `json:"gzip_objects"` @@ -209,7 +208,7 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) { apiReplicationWorkers, apiReplicationFailedWorkers, "expiry_workers", - apiMultipartListing, // Ignore the retired shared-config key during migration. + apiMultipartListing, // Tolerate the retired key without changing other API settings. } disableODirect := env.Get(EnvAPIDisableODirect, kvs.Get(apiDisableODirect)) == config.EnableOn @@ -344,11 +343,5 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) { cfg.ObjectMaxVersions = math.MaxInt64 } - cfg.MultipartListing = env.Get(EnvAPIMultipartListing, "legacy") - if cfg.MultipartListing != "strict" && cfg.MultipartListing != "legacy" { - cfg.MultipartListing = "legacy" - return cfg, fmt.Errorf("%s must be strict or legacy; using legacy", EnvAPIMultipartListing) - } - return cfg, nil } diff --git a/internal/config/api/api_test.go b/internal/config/api/api_test.go index ad8f8697a..96bf6357a 100644 --- a/internal/config/api/api_test.go +++ b/internal/config/api/api_test.go @@ -11,22 +11,21 @@ import ( "github.com/minio/minio/internal/config" ) -func TestMultipartListingMigrationMode(t *testing.T) { +func TestMultipartListingRetiredConfiguration(t *testing.T) { t.Setenv(EnvAPIRequestsMax, "") t.Setenv(EnvAPISyncEvents, "") t.Setenv(EnvAPIObjectMaxVersions, "") for _, tc := range []struct { - name, stored, override, want string - invalid bool + name, stored, override string }{ - {name: "default", want: "legacy"}, - {name: "explicit-legacy", override: "legacy", want: "legacy"}, - {name: "explicit-strict", override: "strict", want: "strict"}, - {name: "retired-stored-strict", stored: "strict", want: "legacy"}, - {name: "retired-stored-legacy", stored: "legacy", want: "legacy"}, - {name: "retired-stored-invalid", stored: "automatic", want: "legacy"}, - {name: "environment-override", stored: "legacy", override: "strict", want: "strict"}, - {name: "invalid-environment", stored: "strict", override: "automatic", want: "legacy", invalid: true}, + {name: "default"}, + {name: "explicit-legacy", override: "legacy"}, + {name: "explicit-strict", override: "strict"}, + {name: "retired-stored-strict", stored: "strict"}, + {name: "retired-stored-legacy", stored: "legacy"}, + {name: "retired-stored-invalid", stored: "automatic"}, + {name: "environment-override", stored: "legacy", override: "strict"}, + {name: "invalid-environment", stored: "strict", override: "automatic"}, } { t.Run(tc.name, func(t *testing.T) { t.Setenv(EnvAPIMultipartListing, tc.override) @@ -38,11 +37,11 @@ func TestMultipartListingMigrationMode(t *testing.T) { kvs.Set(apiMultipartListing, tc.stored) } cfg, err := LookupConfig(kvs) - if (err != nil) != tc.invalid || cfg.MultipartListing != tc.want { - t.Fatalf("mode=%q err=%v, want %q invalid=%v", cfg.MultipartListing, err, tc.want, tc.invalid) + if err != nil { + t.Fatalf("retired setting must be ignored: %v", err) } if cfg.RequestsMax != 17 || !cfg.SyncEvents || cfg.ObjectMaxVersions != 71 { - t.Fatalf("migration setting discarded unrelated config: %+v", cfg) + t.Fatalf("retired setting discarded unrelated config: %+v", cfg) } }) } @@ -58,7 +57,7 @@ func TestMultipartListingConfigPersistence(t *testing.T) { t.Cleanup(func() { config.DefaultKVS = previous }) for _, help := range Help { if help.Key == apiMultipartListing { - t.Fatal("process-only setting advertised as shared config") + t.Fatal("retired setting advertised as shared config") } } for _, input := range []string{ @@ -82,11 +81,11 @@ func TestMultipartListingConfigPersistence(t *testing.T) { } kvs := loaded.Merge()[config.APISubSys][config.Default] if _, present := kvs.Lookup(apiMultipartListing); present { - t.Fatal("process-only setting persisted in shared config") + t.Fatal("retired setting persisted in shared config") } cfg, err := LookupConfig(kvs) - if err != nil || cfg.MultipartListing != "strict" { - t.Fatalf("environment mode lost: %+v %v", cfg, err) + if err != nil { + t.Fatalf("retired environment setting caused failure: %+v %v", cfg, err) } if input != "" && cfg.RequestsMax != 17 { t.Fatalf("unrelated setting lost: %+v", cfg)