Merge pull request #294 from botdeniska/fix/tag-auth-server-error

Preserve server errors for reads authorized by object tags and GetObjectAttributes.

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

cmd/auth-handler-unread-tags.go

@@ -0,0 +1,289 @@
+// Copyright (c) 2026 Denis Babchuk
+//
+// This file is part of Silo Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"context"
+	"encoding/json"
+	"maps"
+	"net/http"
+	"slices"
+	"strings"
+
+	"github.com/minio/minio-go/v7/pkg/set"
+	"github.com/minio/minio-go/v7/pkg/tags"
+	"github.com/minio/minio/internal/auth"
+	"github.com/minio/minio/internal/logger"
+	"github.com/pgsty/silo-pkg/v3/policy"
+	"github.com/pgsty/silo-pkg/v3/policy/condition"
+)
+
+// authorizeRequestWithUnreadTags authorizes action on an object whose stored
+// tags could not be read because reading the object failed with readErr.
+//
+// Without the tags, an Allow statement that tests s3:ExistingObjectTag cannot
+// match, so a caller allowed only by such a statement would get AccessDenied,
+// which clients do not retry, for a transient server error such as lost read
+// quorum. For a server error the request is therefore also authorized as if
+// the object carried tags that the policies ask for: a caller that some tags
+// let through gets the error, a caller that no tags let through still gets
+// AccessDenied. For any other error, NoSuchKey among them, the tags stay empty.
+func authorizeRequestWithUnreadTags(ctx context.Context, r *http.Request, action policy.Action, readErr error) APIErrorCode {
+	s3Err := authorizeRequest(ctx, r, action)
+	if s3Err == ErrNone || toAPIError(ctx, readErr).HTTPStatusCode < http.StatusInternalServerError {
+		return s3Err
+	}
+	for _, existingTags := range policyExistingObjectTags(ctx, r, action) {
+		if authorizeRequestWithExistingTags(ctx, r, action, existingTags) == ErrNone {
+			return ErrNone
+		}
+	}
+	return s3Err
+}
+
+// maxExistingTagCandidates bounds the tag sets tried for one request; each
+// one costs an authorization.
+const maxExistingTagCandidates = 64
+
+// policyExistingObjectTags returns tag sets under which the policies that
+// authorize the request may let action through on its object: the bucket
+// policy for an anonymous request, else the cached IAM policies and, as a
+// second policy set, the session policy of the credential. A tag set passes
+// the s3:ExistingObjectTag tests of one Allow statement of each policy set and,
+// as far as the same tags allow, keeps the Deny statements that test tags from
+// matching. The tag sets are only candidates; the regular authorization
+// decides what they allow.
+func policyExistingObjectTags(ctx context.Context, r *http.Request, action policy.Action) []string {
+	reqInfo := logger.GetReqInfo(ctx)
+	if reqInfo == nil {
+		return nil
+	}
+	bucket, object := reqInfo.BucketName, reqInfo.ObjectName
+	var (
+		allows [][]map[string]string // per policy set, the tags its Allow statements ask for
+		denies []map[string]string   // tags that keep a Deny statement from matching
+	)
+	// add files the tags of a statement that applies to action on the
+	// object, whatever its conditions.
+	add := func(effect policy.Effect, conditions condition.Functions, values map[string][]string, fromSet *[]map[string]string) {
+		if tagMap, ok := existingObjectTagsFor(effect, conditions, values); ok {
+			if effect == policy.Allow {
+				*fromSet = append(*fromSet, tagMap)
+			} else {
+				denies = append(denies, tagMap)
+			}
+		}
+	}
+	switch {
+	case reqInfo.Cred.AccessKey == "":
+		p, err := globalPolicySys.Get(bucket)
+		if err != nil {
+			return nil
+		}
+		values := getConditionValues(r, reqInfo.Region, auth.AnonymousCredentials)
+		var fromSet []map[string]string
+		for _, st := range p.Statements {
+			match := st
+			match.Effect, match.Conditions = policy.Allow, nil
+			if match.IsAllowed(policy.BucketPolicyArgs{Action: action, BucketName: bucket, ObjectName: object, ConditionValues: values}) {
+				add(st.Effect, st.Conditions, values, &fromSet)
+			}
+		}
+		allows = append(allows, fromSet)
+	case newGlobalAuthZPluginFn() == nil && globalIAMSys.Initialized():
+		values := getConditionValues(r, "", reqInfo.Cred)
+		// From the cache only: reading the drives may be what just failed.
+		docs, _ := globalIAMSys.store.listPolicyDocs(ctx, "")
+		policySets := [][]policy.Policy{nil}
+		for _, name := range slices.Sorted(maps.Keys(docs)) {
+			policySets[0] = append(policySets[0], docs[name].Policy)
+		}
+		if sp, ok := reqInfo.Cred.Claims[sessionPolicyNameExtracted].(string); ok {
+			if p, err := policy.ParseConfig(strings.NewReader(sp)); err == nil {
+				policySets = append(policySets, []policy.Policy{*p})
+			}
+		}
+		for _, policies := range policySets {
+			var fromSet []map[string]string
+			for _, p := range policies {
+				for _, st := range p.Statements {
+					match := st
+					match.Effect, match.Conditions = policy.Allow, nil
+					if match.IsAllowed(policy.Args{Action: action, BucketName: bucket, ObjectName: object, ConditionValues: values}) {
+						add(st.Effect, st.Conditions, values, &fromSet)
+					}
+				}
+			}
+			allows = append(allows, fromSet)
+		}
+	}
+	return combineExistingObjectTags(allows, denies)
+}
+
+// combineExistingObjectTags merges the tags of one Allow statement of each
+// policy set, or of none for a set that may allow without tags, and tries each
+// result also with the tags that keep the Deny statements from matching. The
+// first tags win a conflict; the authorization of the result settles it.
+func combineExistingObjectTags(allows [][]map[string]string, denies []map[string]string) []string {
+	merge := func(parts ...map[string]string) map[string]string {
+		merged := make(map[string]string)
+		for _, part := range parts {
+			for k, v := range part {
+				if _, ok := merged[k]; !ok {
+					merged[k] = v
+				}
+			}
+		}
+		return merged
+	}
+	combos := [][]map[string]string{nil}
+	for _, fromSet := range allows {
+		// Many statements ask for the same tags.
+		seen := set.NewStringSet()
+		fromSet = slices.DeleteFunc(slices.Clone(fromSet), func(tagMap map[string]string) bool {
+			t, err := tags.MapToObjectTags(tagMap)
+			if err != nil || seen.Contains(t.String()) {
+				return true
+			}
+			seen.Add(t.String())
+			return false
+		})
+		next := make([][]map[string]string, 0, len(combos)*(len(fromSet)+1))
+		for _, combo := range combos {
+			next = append(next, combo)
+			for _, tagMap := range fromSet {
+				next = append(next, append(slices.Clip(combo), tagMap))
+			}
+		}
+		combos = next[:min(len(next), maxExistingTagCandidates)]
+	}
+	deny := merge(denies...)
+	found := set.NewStringSet()
+	var candidates []string
+	for _, combo := range combos {
+		for _, tagMap := range []map[string]string{merge(combo...), merge(append(slices.Clip(combo), deny)...)} {
+			t, err := tags.MapToObjectTags(tagMap)
+			if len(tagMap) == 0 || err != nil || found.Contains(t.String()) {
+				continue
+			}
+			found.Add(t.String())
+			if candidates = append(candidates, t.String()); len(candidates) == maxExistingTagCandidates {
+				return candidates
+			}
+		}
+	}
+	return candidates
+}
+
+// existingObjectTagsFor returns tags under which the s3:ExistingObjectTag tests
+// of conditions all pass, for an Allow statement, or one of them fails, for a
+// Deny statement; a tag left out is absent. It returns false if the conditions
+// test no tag, or if no such tags were found among those tried for each tag:
+// none, the values the tests name, with policy variables taken from values and
+// wildcards filled, and the fillers "" and "_". The policy engine itself
+// evaluates the tests for each of them.
+func existingObjectTagsFor(effect policy.Effect, conditions condition.Functions, values map[string][]string) (map[string]string, bool) {
+	// Each function tests one key, so the tests of one tag pass or fail
+	// whatever the other tags are.
+	byKey := make(map[string]condition.Functions)
+	for _, f := range conditions {
+		for key := range condition.NewFunctions(f).Keys() {
+			if key.Is(condition.ExistingObjectTag) {
+				byKey[key.Name()] = append(byKey[key.Name()], f)
+			}
+		}
+	}
+	if len(byKey) == 0 {
+		return nil, false
+	}
+	tagMap := make(map[string]string)
+	for _, key := range slices.Sorted(maps.Keys(byKey)) {
+		tagKey := strings.TrimPrefix(key, condition.ExistingObjectTag.Name()+"/")
+		value, present, found := existingObjectTagValue(key, tagKey, byKey[key], values, effect == policy.Allow)
+		switch {
+		case effect == policy.Deny && found:
+			if present {
+				tagMap[tagKey] = value
+			}
+			return tagMap, true
+		case effect == policy.Allow && !found:
+			return nil, false
+		case present:
+			tagMap[tagKey] = value
+		}
+	}
+	if effect == policy.Deny {
+		return nil, false
+	}
+	return tagMap, true
+}
+
+// existingObjectTagValue looks for a value of tag tagKey, or its absence,
+// under which tests, the functions that test the condition key key, evaluate
+// to want.
+func existingObjectTagValue(key, tagKey string, tests condition.Functions, values map[string][]string, want bool) (value string, present, found bool) {
+	defer delete(values, key)
+	delete(values, key)
+	if tests.Evaluate(values) == want {
+		return "", false, true
+	}
+	// The functions keep their values unexported; JSON exposes them.
+	buf, err := json.Marshal(tests)
+	if err != nil {
+		return "", false, false
+	}
+	var byName map[string]map[string]condition.ValueSet
+	if err = json.Unmarshal(buf, &byName); err != nil {
+		return "", false, false
+	}
+	var tried []string
+	for name, byKey := range byName {
+		if name == "Null" {
+			continue // its operand says whether the tag is there, not a value
+		}
+		for _, valueSet := range byKey {
+			for _, v := range valueSet.ToSlice() {
+				s := substituteConditionVariables(v.String(), values)
+				tried = append(tried, s, strings.NewReplacer("*", "", "?", "_").Replace(s))
+			}
+		}
+	}
+	slices.Sort(tried)
+	for _, v := range append(slices.Compact(tried), "", "_") {
+		if _, err := tags.MapToObjectTags(map[string]string{tagKey: v}); err != nil {
+			continue
+		}
+		values[key] = []string{v}
+		if tests.Evaluate(values) == want {
+			return v, true, true
+		}
+	}
+	return "", false, false
+}
+
+// substituteConditionVariables resolves the policy variables in v from the
+// condition values as the policy engine does for the values of a condition;
+// silo-pkg keeps its own function for this unexported.
+func substituteConditionVariables(v string, values map[string][]string) string {
+	for _, key := range condition.CommonKeys {
+		if rvalues, ok := values[key.Name()]; ok && len(rvalues) > 0 && rvalues[0] != "" {
+			v = strings.ReplaceAll(v, key.VarName(), rvalues[0])
+		}
+	}
+	return v
+}

cmd/auth-handler-unread-tags_test.go

@@ -0,0 +1,305 @@
+// Copyright (c) 2026 Denis Babchuk
+//
+// This file is part of Silo Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"encoding/xml"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/minio/minio-go/v7/pkg/tags"
+	"github.com/minio/minio/internal/auth"
+	xhttp "github.com/minio/minio/internal/http"
+	"github.com/pgsty/silo-pkg/v3/policy"
+	"github.com/pgsty/silo-pkg/v3/policy/condition"
+)
+
+// objectReadFault takes the first n drives of the set that holds object
+// offline for the metadata reads of that object only, so that IAM and bucket
+// metadata stay readable. It returns a function that restores the drives.
+func objectReadFault(z *erasureServerPools, bucket, object string, n int) func() {
+	pool := z.serverPools[0]
+	setIndex := pool.getHashedSetIndex(object)
+	pool.erasureDisksMu.Lock()
+	defer pool.erasureDisksMu.Unlock()
+	previous := pool.erasureDisks[setIndex]
+	disks := append([]StorageAPI(nil), previous...)
+	for i := range n {
+		disks[i] = consistencyReadFaultDisk{StorageAPI: disks[i], bucket: bucket, object: object}
+	}
+	pool.erasureDisks[setIndex] = disks
+	return func() {
+		pool.erasureDisksMu.Lock()
+		pool.erasureDisks[setIndex] = previous
+		pool.erasureDisksMu.Unlock()
+	}
+}
+
+// A caller allowed to read an object only through s3:ExistingObjectTag must
+// get the storage error, which clients retry, and not AccessDenied, when the
+// object's tags cannot be read because read quorum is lost. A caller that no
+// tags would let through must still get AccessDenied, and nothing changes
+// while the tags can be read.
+func TestExistingObjectTagAuthReadQuorumLost(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	obj, dirs, err := prepareErasure16(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	z := obj.(*erasureServerPools)
+	previous := newObjectLayerFn()
+	setObjectLayer(z)
+	defer func() {
+		z.Shutdown(context.Background())
+		removeRoots(dirs)
+		setObjectLayer(previous)
+	}()
+	bucket, router, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	statement := func(effect, resource, conditions string) string {
+		st := fmt.Sprintf(`{"Effect": %q, "Action": ["s3:GetObject", "s3:PutObjectTagging"], "Resource": ["arn:aws:s3:::%s/%s"]`,
+			effect, bucket, resource)
+		if conditions != "" {
+			st += `, "Condition": ` + conditions
+		}
+		return st + "}"
+	}
+	tagStatement := func(effect, resource, value string) string {
+		return statement(effect, resource, fmt.Sprintf(`{"StringEquals": {"s3:ExistingObjectTag/k": %q}}`, value))
+	}
+	newUser := func(statements ...string) auth.Credentials {
+		return newDeleteAuthzPolicyUser(t, "", bucket, "["+strings.Join(statements, ",")+"]")
+	}
+	newServiceAccount := func(parent auth.Credentials, statements string) auth.Credentials {
+		sessionPolicy, err := policy.ParseConfig(strings.NewReader(`{"Version": "2012-10-17", "Statement": [` + statements + `]}`))
+		if err != nil {
+			t.Fatal(err)
+		}
+		cred, _, err := globalIAMSys.NewServiceAccount(ctx, parent.AccessKey, nil, newServiceAccountOpts{sessionPolicy: sessionPolicy})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return cred
+	}
+
+	type user struct {
+		name  string
+		creds auth.Credentials
+		// The status with the object's tags readable and with read quorum
+		// lost: 200 and 503 for a caller that some tags let through, 403
+		// and 403 for a caller that none do.
+		healthy, quorumLost int
+	}
+	allowedByTags := func(name string, creds auth.Credentials) user {
+		return user{name, creds, http.StatusOK, http.StatusServiceUnavailable}
+	}
+	deniedByAnyTags := func(name string, creds auth.Credentials) user {
+		return user{name, creds, http.StatusForbidden, http.StatusForbidden}
+	}
+	// Each case below that tests a tag key of its own uses a key no other
+	// policy asks for.
+	ownerCondition := `{"StringEquals": {"s3:ExistingObjectTag/owner": "${aws:username}"}}`
+	owner := newUser(statement("Allow", "*", ownerCondition))
+	users := []user{
+		// Allowed only while the object carries k=v.
+		allowedByTags("tagged", newUser(tagStatement("Allow", "*", "v"))),
+		// Allowed while the object carries a tag np, whatever its value.
+		allowedByTags("nullFalse", newUser(statement("Allow", "*", `{"Null": {"s3:ExistingObjectTag/np": "false"}}`))),
+		// Allowed while a tag ne is there and is not w.
+		allowedByTags("notEquals", newUser(statement("Allow", "*",
+			`{"Null": {"s3:ExistingObjectTag/ne": "false"}, "StringNotEquals": {"s3:ExistingObjectTag/ne": "w"}}`))),
+		// Allowed while every value of tag fa differs from w; that needs a tag fa.
+		allowedByTags("forAllNotEquals", newUser(statement("Allow", "*", `{"ForAllValues:StringNotEquals": {"s3:ExistingObjectTag/fa": "w"}}`))),
+		// Allowed while the object is tagged with the caller's name.
+		allowedByTags("owner", owner),
+		// Allowed, but denied unless the object carries du=1.
+		allowedByTags("denyUnlessTagged", newUser(statement("Allow", "*", ""),
+			statement("Deny", "*", `{"StringNotEquals": {"s3:ExistingObjectTag/du": "1"}}`))),
+		// A service account whose parent asks for k=v and whose own policy
+		// asks for j=w: only an object carrying both lets it through.
+		allowedByTags("serviceAccount", newServiceAccount(newUser(tagStatement("Allow", "*", "v")),
+			statement("Allow", "*", `{"StringEquals": {"s3:ExistingObjectTag/j": "w"}}`))),
+
+		// Allowed only for k=w: the decision still depends on the tags.
+		{"otherValue", newUser(tagStatement("Allow", "*", "w")), http.StatusForbidden, http.StatusServiceUnavailable},
+
+		// Allowed for k=v under another prefix only.
+		deniedByAnyTags("otherPrefix", newUser(tagStatement("Allow", "other/*", "v"))),
+		// Allowed and denied for k=v alike.
+		deniedByAnyTags("allowDeny", newUser(tagStatement("Allow", "*", "v"), tagStatement("Deny", "*", "v"))),
+		// Allowed only without a tag nc that is 1.
+		deniedByAnyTags("nullConflict", newUser(statement("Allow", "*",
+			`{"Null": {"s3:ExistingObjectTag/nc": "true"}, "StringEquals": {"s3:ExistingObjectTag/nc": "1"}}`))),
+		// Tagged with the caller's name, under another prefix only.
+		deniedByAnyTags("ownerPrefix", newUser(statement("Allow", "other/*", ownerCondition))),
+		// Denied unless the object carries db=1, and denied if it does.
+		deniedByAnyTags("denyBoth", newUser(statement("Allow", "*", ""),
+			statement("Deny", "*", `{"StringNotEquals": {"s3:ExistingObjectTag/db": "1"}}`),
+			statement("Deny", "*", `{"StringEquals": {"s3:ExistingObjectTag/db": "1"}}`))),
+		// The parent asks for k=v and the service account for k=x.
+		deniedByAnyTags("serviceAccountConflict", newServiceAccount(newUser(tagStatement("Allow", "*", "v")),
+			tagStatement("Allow", "*", "x"))),
+	}
+
+	const object, missing = "tagged", "missing"
+	if _, err := z.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""),
+		ObjectOptions{UserDefined: map[string]string{xhttp.AmzObjectTagging: "k=v&j=w&np=1&ne=1&fa=1&du=1&owner=" + owner.AccessKey}}); err != nil {
+		t.Fatal(err)
+	}
+	// Anonymous reads are allowed by the bucket policy while the object carries k=v.
+	if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketPolicyConfig, []byte(`{"Version": "2012-10-17", "Statement": [
+  {"Effect": "Allow", "Principal": {"AWS": ["*"]}, "Action": ["s3:GetObject"], "Resource": ["arn:aws:s3:::`+bucket+`/*"],
+   "Condition": {"StringEquals": {"s3:ExistingObjectTag/k": "v"}}}]}`)); err != nil {
+		t.Fatal(err)
+	}
+
+	type request struct {
+		name           string
+		method, target string
+		header         map[string]string
+		body           []byte
+	}
+	objectURL := getGetObjectURL("", bucket, object)
+	get := request{"GET", http.MethodGet, objectURL, nil, nil}
+	head := request{"HEAD", http.MethodHead, objectURL, nil, nil}
+	taggingURL := makeTestTargetURL("", bucket, object, url.Values{"tagging": []string{""}})
+	taggingBody := []byte(`<Tagging><TagSet><Tag><Key>k</Key><Value>v</Value></Tag></TagSet></Tagging>`)
+	putTagging := request{"PutObjectTagging", http.MethodPut, taggingURL, nil, taggingBody}
+	check := func(stage string, creds auth.Credentials, name string, req request, want int) {
+		t.Helper()
+		var rec *httptest.ResponseRecorder
+		if creds.AccessKey == "" {
+			anonymous, err := newTestRequest(req.method, req.target, 0, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			rec = httptest.NewRecorder()
+			router.ServeHTTP(rec, anonymous)
+		} else {
+			var body io.ReadSeeker
+			if req.body != nil {
+				body = bytes.NewReader(req.body)
+			}
+			signed, err := newTestSignedRequestV4(req.method, req.target, int64(len(req.body)), body, creds.AccessKey, creds.SecretKey, req.header)
+			if err != nil {
+				t.Fatal(err)
+			}
+			rec = httptest.NewRecorder()
+			router.ServeHTTP(rec, signed)
+		}
+		if rec.Code != want {
+			var apiErr APIErrorResponse
+			_ = xml.Unmarshal(rec.Body.Bytes(), &apiErr)
+			t.Errorf("%s: %s by %s: status %d %s, want %d", stage, req.name, name, rec.Code, apiErr.Code, want)
+		}
+	}
+
+	var anonymous auth.Credentials
+	for _, u := range users {
+		check("healthy", u.creds, u.name, get, u.healthy)
+	}
+	check("healthy", anonymous, "anonymous", get, http.StatusOK)
+	// The tags were read and allow the request: the range error is the answer.
+	check("healthy", users[0].creds, users[0].name,
+		request{"GET bytes=100-200", http.MethodGet, objectURL, map[string]string{xhttp.Range: "bytes=100-200"}, nil},
+		http.StatusRequestedRangeNotSatisfiable)
+	// Without an object there are no tags: AccessDenied, as before.
+	missingURL := getGetObjectURL("", bucket, missing)
+	check("missing key", users[0].creds, users[0].name, request{"GET", http.MethodGet, missingURL, nil, nil}, http.StatusForbidden)
+	check("missing key", users[0].creds, users[0].name, request{"HEAD", http.MethodHead, missingURL, nil, nil}, http.StatusForbidden)
+
+	defer objectReadFault(z, bucket, object, 10)()
+	if _, err := z.GetObjectInfo(ctx, bucket, object, ObjectOptions{}); !isErrReadQuorum(err) {
+		t.Fatalf("GetObjectInfo with 10 of 16 drives failing: %v, want InsufficientReadQuorum", err)
+	}
+	for _, req := range []request{get, head, putTagging} {
+		for _, u := range users {
+			check("read quorum lost", u.creds, u.name, req, u.quorumLost)
+		}
+	}
+	check("read quorum lost", anonymous, "anonymous", get, http.StatusServiceUnavailable)
+	check("read quorum lost", anonymous, "anonymous", head, http.StatusServiceUnavailable)
+}
+
+func TestExistingObjectTagsFor(t *testing.T) {
+	values := map[string][]string{"username": {"alice"}, "sub": {"u1"}}
+	for _, tc := range []struct {
+		effect     policy.Effect
+		conditions string
+		want       string // the tags to set, encoded as in x-amz-tagging
+		ok         bool   // false: no tags for this statement
+	}{
+		{policy.Allow, `{"StringEquals": {"s3:ExistingObjectTag/k": "v"}}`, "k=v", true},
+		{policy.Allow, `{"StringEqualsIgnoreCase": {"s3:ExistingObjectTag/k": "V"}, "IpAddress": {"aws:SourceIp": "10.0.0.0/8"}}`, "k=V", true},
+		{policy.Allow, `{"ForAnyValue:StringEquals": {"s3:ExistingObjectTag/k": ["b", "a"]}}`, "k=a", true},
+		{policy.Allow, `{"StringLike": {"s3:ExistingObjectTag/k": "a?c*"}}`, "k=a_c", true},
+		{policy.Allow, `{"StringEquals": {"s3:ExistingObjectTag/b": "2", "s3:ExistingObjectTag/a": "1"}}`, "a=1&b=2", true},
+		// Passed without the tag.
+		{policy.Allow, `{"StringNotEquals": {"s3:ExistingObjectTag/k": "v"}}`, "", true},
+		{policy.Allow, `{"Null": {"s3:ExistingObjectTag/k": "true"}}`, "", true},
+		// Null "false" asks for the tag, with any value.
+		{policy.Allow, `{"Null": {"s3:ExistingObjectTag/k": "false"}}`, "k=", true},
+		{policy.Allow, `{"Null": {"s3:ExistingObjectTag/k": "false"}, "StringNotEquals": {"s3:ExistingObjectTag/k": ["", "w"]}}`, "k=_", true},
+		{policy.Allow, `{"Null": {"s3:ExistingObjectTag/k": "false"}, "StringNotLike": {"s3:ExistingObjectTag/k": "*w"}}`, "k=", true},
+		{policy.Allow, `{"ForAllValues:StringNotEquals": {"s3:ExistingObjectTag/k": "w"}}`, "k=", true},
+		{policy.Allow, `{"Null": {"s3:ExistingObjectTag/k": "true"}, "StringEquals": {"s3:ExistingObjectTag/k": "v"}}`, "", false},
+		// Policy variables take the values of the request.
+		{policy.Allow, `{"StringEquals": {"s3:ExistingObjectTag/owner": "${aws:username}"}}`, "owner=alice", true},
+		{policy.Allow, `{"StringLike": {"s3:ExistingObjectTag/owner": "${jwt:sub}-*"}}`, "owner=u1-", true},
+		{policy.Allow, `{"StringEquals": {"s3:ExistingObjectTag/owner": "${ldap:user}"}}`, "", false},
+		// A Deny is kept from matching.
+		{policy.Deny, `{"StringNotEquals": {"s3:ExistingObjectTag/k": "v"}}`, "k=v", true},
+		{policy.Deny, `{"Null": {"s3:ExistingObjectTag/k": "true"}}`, "k=", true},
+		{policy.Deny, `{"StringEquals": {"s3:ExistingObjectTag/k": "v"}}`, "", true},
+		{policy.Deny, `{"StringEquals": {"s3:prefix": "v"}}`, "", false},
+		// No tag tests.
+		{policy.Allow, `{"StringEquals": {"s3:prefix": "v"}}`, "", false},
+		{policy.Allow, "", "", false},
+	} {
+		var conditions condition.Functions // nil: a statement without a Condition
+		if tc.conditions != "" {
+			if err := json.Unmarshal([]byte(tc.conditions), &conditions); err != nil {
+				t.Fatalf("%s: %v", tc.conditions, err)
+			}
+		}
+		tagMap, ok := existingObjectTagsFor(tc.effect, conditions, values)
+		got := ""
+		if len(tagMap) > 0 {
+			encoded, err := tags.MapToObjectTags(tagMap)
+			if err != nil {
+				t.Fatalf("%s: %v", tc.conditions, err)
+			}
+			got = encoded.String()
+		}
+		if got != tc.want || ok != tc.ok {
+			t.Errorf("%s %s: got %q %v, want %q %v", tc.effect, tc.conditions, got, ok, tc.want, tc.ok)
+		}
+	}
+}

cmd/auth-handler.go

@@ -425,6 +425,19 @@ func authorizeRequest(ctx context.Context, r *http.Request, action policy.Action
 	return authorizeRequestWithExistingTags(ctx, r, action, "")
 }
 
+// authorizeBucketRequest authorizes a bucket action, such as s3:ListBucket,
+// on the bucket of a request made for one of its objects.
+func authorizeBucketRequest(ctx context.Context, r *http.Request, action policy.Action) APIErrorCode {
+	reqInfo := logger.GetReqInfo(ctx)
+	if reqInfo == nil {
+		return ErrAccessDenied
+	}
+	objectName := reqInfo.ObjectName
+	reqInfo.ObjectName = ""
+	defer func() { reqInfo.ObjectName = objectName }()
+	return authorizeRequest(ctx, r, action)
+}
+
 func deleteObjectAction(versionID string) policy.Action {
 	if versionID != "" {
 		return policy.DeleteObjectVersionAction

cmd/object-attributes-read-error_test.go

@@ -0,0 +1,91 @@
+// Copyright (c) 2026 Denis Babchuk
+//
+// This file is part of Silo Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"encoding/xml"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+
+	"github.com/minio/minio/internal/auth"
+	xhttp "github.com/minio/minio/internal/http"
+)
+
+// GetObjectAttributes answers NoSuchKey with AccessDenied to a caller without
+// s3:ListBucket on the bucket, as S3 does, so that a missing object cannot be
+// told apart from a forbidden one. A server error, such as lost read quorum,
+// must reach a caller allowed to read the object as the error it is.
+func TestGetObjectAttributesReadQuorumLost(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	obj, dirs, err := prepareErasure16(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	z := obj.(*erasureServerPools)
+	previous := newObjectLayerFn()
+	setObjectLayer(z)
+	defer func() {
+		z.Shutdown(context.Background())
+		removeRoots(dirs)
+		setObjectLayer(previous)
+	}()
+	bucket, router, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	const object, missing = "object", "missing"
+	if _, err := z.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+
+	read := `{"Effect": "Allow", "Action": ["s3:GetObject", "s3:GetObjectAttributes"], "Resource": ["arn:aws:s3:::` + bucket + `/*"]}`
+	withoutList := newDeleteAuthzPolicyUser(t, "", bucket, "["+read+"]")
+	withList := newDeleteAuthzPolicyUser(t, "", bucket, "["+read+
+		`, {"Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": ["arn:aws:s3:::`+bucket+`"]}]`)
+
+	check := func(stage, user string, creds auth.Credentials, key string, want int) {
+		t.Helper()
+		target := makeTestTargetURL("", bucket, key, url.Values{"attributes": []string{""}})
+		req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, creds.AccessKey, creds.SecretKey,
+			map[string]string{xhttp.AmzObjectAttributes: "ETag"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		rec := httptest.NewRecorder()
+		router.ServeHTTP(rec, req)
+		if rec.Code != want {
+			var apiErr APIErrorResponse
+			_ = xml.Unmarshal(rec.Body.Bytes(), &apiErr)
+			t.Errorf("%s: GetObjectAttributes %s by %s: status %d %s, want %d", stage, key, user, rec.Code, apiErr.Code, want)
+		}
+	}
+
+	check("healthy", "withoutList", withoutList, object, http.StatusOK)
+	check("healthy", "withList", withList, object, http.StatusOK)
+	check("healthy", "withoutList", withoutList, missing, http.StatusForbidden)
+	check("healthy", "withList", withList, missing, http.StatusNotFound)
+
+	defer objectReadFault(z, bucket, object, 10)()
+	check("read quorum lost", "withoutList", withoutList, object, http.StatusServiceUnavailable)
+	check("read quorum lost", "withList", withList, object, http.StatusServiceUnavailable)
+}

cmd/object-handlers.go

@@ -395,12 +395,14 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj
 	}
 
 	// Validate pre-conditions if any.
+	var storedTags *string // set once the object has been read far enough
 	opts.CheckPrecondFn = func(oi ObjectInfo) bool {
 		if _, err := DecryptObjectInfo(&oi, r); err != nil {
 			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
 			return true
 		}
 
+		storedTags = &oi.UserTags
 		if s3Error := authorizeRequestWithExistingTags(ctx, r, policy.GetObjectAction, oi.UserTags); s3Error != ErrNone {
 			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
 			return true
@@ -445,8 +447,14 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj
 				return
 			}
 			// validate if the request indeed was authorized, if it wasn't we need to return "ErrAccessDenied"
-			// instead of any namespace related error.
-			if s3Error := authorizeRequest(ctx, r, policy.GetObjectAction); s3Error != ErrNone {
+			// instead of any namespace related error. Use the stored tags if the callback read them.
+			var s3Error APIErrorCode
+			if storedTags != nil {
+				s3Error = authorizeRequestWithExistingTags(ctx, r, policy.GetObjectAction, *storedTags)
+			} else {
+				s3Error = authorizeRequestWithUnreadTags(ctx, r, policy.GetObjectAction, err)
+			}
+			if s3Error != ErrNone {
 				writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
 				return
 			}
@@ -623,12 +631,15 @@ func (api objectAPIHandlers) getObjectAttributesHandler(ctx context.Context, obj
 
 	objInfo, err := objectAPI.GetObjectInfo(ctx, bucket, object, opts)
 	if err != nil {
-		s3Error = authorizeRequest(ctx, r, policy.ListBucketAction)
-		if s3Error == ErrNone {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-			return
+		// As with S3, a caller without s3:ListBucket on the bucket gets
+		// AccessDenied in place of NoSuchKey. A server error, such as lost
+		// read quorum, tells nothing about the object and reaches the caller,
+		// who may read the object, as it is.
+		apiErr := toAPIError(ctx, err)
+		if apiErr.HTTPStatusCode < http.StatusInternalServerError && authorizeBucketRequest(ctx, r, policy.ListBucketAction) != ErrNone {
+			apiErr = errorCodes.ToAPIErr(ErrAccessDenied)
 		}
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
+		writeErrorResponse(ctx, w, apiErr, r.URL)
 		return
 	}
 
@@ -925,7 +936,14 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
 		}
 	}
 
-	if s3Error := authorizeRequestWithExistingTags(ctx, r, policy.GetObjectAction, objInfo.UserTags); s3Error != ErrNone {
+	// objInfo carries the stored tags unless reading the object failed.
+	var s3Error APIErrorCode
+	if err != nil && !proxy.Proxy {
+		s3Error = authorizeRequestWithUnreadTags(ctx, r, policy.GetObjectAction, err)
+	} else {
+		s3Error = authorizeRequestWithExistingTags(ctx, r, policy.GetObjectAction, objInfo.UserTags)
+	}
+	if s3Error != ErrNone {
 		writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(s3Error))
 		return
 	}
@@ -3783,11 +3801,13 @@ func (api objectAPIHandlers) PutObjectTaggingHandler(w http.ResponseWriter, r *h
 	}
 
 	objInfo, err := objAPI.GetObjectInfo(ctx, bucket, object, opts)
-	existingTags := ""
-	if err == nil {
-		existingTags = objInfo.UserTags
+	var s3Error APIErrorCode
+	if err != nil {
+		s3Error = authorizeRequestWithUnreadTags(ctx, r, policy.PutObjectTaggingAction, err)
+	} else {
+		s3Error = authorizeRequestWithExistingTags(ctx, r, policy.PutObjectTaggingAction, objInfo.UserTags)
 	}
-	if s3Error := authorizeRequestWithExistingTags(ctx, r, policy.PutObjectTaggingAction, existingTags); s3Error != ErrNone {
+	if s3Error != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
 		return
 	}