diff --git a/cmd/auth-handler-unread-tags.go b/cmd/auth-handler-unread-tags.go
index a2db1bc8b..b3b95f894 100644
--- a/cmd/auth-handler-unread-tags.go
+++ b/cmd/auth-handler-unread-tags.go
@@ -50,6 +50,24 @@ func authorizeRequestWithUnreadTags(ctx context.Context, r *http.Request, action
if s3Err == ErrNone || toAPIError(ctx, readErr).HTTPStatusCode < http.StatusInternalServerError {
return s3Err
}
+ return authorizeRequestWithTagCandidates(ctx, r, action, s3Err)
+}
+
+// authorizeRequestWithPotentialTags performs the authorization needed before
+// a remote replica can be read. An empty tag set is enough to reject a caller
+// with no possible access, but a caller whose Allow depends on existing tags
+// must be admitted so the remote copy can supply the authoritative tags.
+// The returned decision is only a gate; the actual tags must be authorized
+// again after they are read.
+func authorizeRequestWithPotentialTags(ctx context.Context, r *http.Request, action policy.Action) APIErrorCode {
+ s3Err := authorizeRequest(ctx, r, action)
+ if s3Err == ErrNone {
+ return ErrNone
+ }
+ return authorizeRequestWithTagCandidates(ctx, r, action, s3Err)
+}
+
+func authorizeRequestWithTagCandidates(ctx context.Context, r *http.Request, action policy.Action, s3Err APIErrorCode) APIErrorCode {
for _, existingTags := range policyExistingObjectTags(ctx, r, action) {
if authorizeRequestWithExistingTags(ctx, r, action, existingTags) == ErrNone {
return ErrNone
diff --git a/cmd/auth-handler-unread-tags_test.go b/cmd/auth-handler-unread-tags_test.go
index aa3f1e822..2a4a4af4d 100644
--- a/cmd/auth-handler-unread-tags_test.go
+++ b/cmd/auth-handler-unread-tags_test.go
@@ -87,7 +87,7 @@ func TestExistingObjectTagAuthReadQuorumLost(t *testing.T) {
}
statement := func(effect, resource, conditions string) string {
- st := fmt.Sprintf(`{"Effect": %q, "Action": ["s3:GetObject", "s3:PutObjectTagging"], "Resource": ["arn:aws:s3:::%s/%s"]`,
+ st := fmt.Sprintf(`{"Effect": %q, "Action": ["s3:GetObject", "s3:GetObjectTagging", "s3:PutObjectTagging", "s3:DeleteObjectTagging"], "Resource": ["arn:aws:s3:::%s/%s"]`,
effect, bucket, resource)
if conditions != "" {
st += `, "Condition": ` + conditions
@@ -263,6 +263,8 @@ func TestExistingObjectTagAuthReadQuorumLost(t *testing.T) {
taggingURL := makeTestTargetURL("", bucket, object, url.Values{"tagging": []string{""}})
taggingBody := []byte(`kv`)
putTagging := request{"PutObjectTagging", http.MethodPut, taggingURL, nil, taggingBody}
+ getTagging := request{"GetObjectTagging", http.MethodGet, taggingURL, nil, nil}
+ deleteTagging := request{"DeleteObjectTagging", http.MethodDelete, taggingURL, nil, nil}
check := func(stage string, creds auth.Credentials, name string, req request, want int) {
t.Helper()
var rec *httptest.ResponseRecorder
@@ -302,6 +304,7 @@ func TestExistingObjectTagAuthReadQuorumLost(t *testing.T) {
var anonymous auth.Credentials
for _, u := range users {
check("healthy", u.creds, u.name, get, u.healthy)
+ check("healthy", u.creds, u.name, getTagging, u.healthy)
}
check("healthy", anonymous, "anonymous", get, http.StatusOK)
// The tags were read and allow the request: the range error is the answer.
@@ -312,12 +315,13 @@ func TestExistingObjectTagAuthReadQuorumLost(t *testing.T) {
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)
+ check("missing key", users[0].creds, users[0].name, request{"GetObjectTagging", http.MethodGet, missingURL + "?tagging", 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 _, req := range []request{get, head, getTagging, putTagging, deleteTagging} {
for _, u := range users {
check("read quorum lost", u.creds, u.name, req, u.quorumLost)
}
diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go
index 81099f504..fb942d2c0 100644
--- a/cmd/bucket-replication.go
+++ b/cmd/bucket-replication.go
@@ -2773,67 +2773,107 @@ func scheduleReplication(ctx context.Context, oi ObjectInfo, o ObjectLayer, dsc
}
}
-// proxyTaggingToRepTarget proxies tagging requests to remote targets for
-// active-active replicated setups
-func proxyTaggingToRepTarget(ctx context.Context, bucket, object string, tags *tags.Tags, opts ObjectOptions, proxyTargets *madmin.BucketTargets, authorize func(string, error) APIErrorCode) (proxy proxyResult, authErr APIErrorCode) {
- if opts.ProxyRequest || (opts.ProxyHeaderSet && !opts.ProxyRequest) {
- return proxy, ErrNone
- }
- // Keep the exact clients whose addressed versions we read. A second
- // discovery could include a newly available, unauthorized destination.
- var targets []*TargetClient
+// taggingProxyTarget keeps the exact client and tags that were authorized.
+// A second discovery must not introduce an unauthorized destination.
+type taggingProxyTarget struct {
+ client *TargetClient
+ tags *tags.Tags
+}
+
+// readTaggingProxyTargets authorizes all present replicas before returning any
+// tags or allowing any mutation. Missing replicas may simply lag replication;
+// unreadable replicas must still fail closed.
+func readTaggingProxyTargets(ctx context.Context, bucket, object string, opts ObjectOptions, proxyTargets *madmin.BucketTargets, authorize func(string, error) APIErrorCode) ([]taggingProxyTarget, proxyResult, APIErrorCode) {
+ var targets []taggingProxyTarget
for _, target := range proxyTargets.Targets {
tgt := globalBucketTargetSys.GetRemoteTargetClient(bucket, target.Arn)
if tgt == nil || globalBucketTargetSys.isOffline(tgt.EndpointURL()) || tgt.disableProxy {
continue
}
- targets = append(targets, tgt)
+ targets = append(targets, taggingProxyTarget{client: tgt})
}
if len(targets) == 0 {
err := RemoteTargetConnectionErr{Bucket: bucket, Err: errors.New("no available remote replication target")}
- return proxyResult{Err: err}, authorize("", err)
+ return nil, proxyResult{Err: err}, authorize("", err)
}
var wg sync.WaitGroup
errs := make([]error, len(targets))
- existingTags := make([]string, len(targets))
- for i, tgt := range targets {
+ for i := range targets {
wg.Go(func() {
- remoteTags, err := tgt.GetObjectTagging(ctx, tgt.Bucket, object, minio.GetObjectTaggingOptions{
+ tgt := &targets[i]
+ tgt.tags, errs[i] = tgt.client.GetObjectTagging(ctx, tgt.client.Bucket, object, minio.GetObjectTaggingOptions{
VersionID: opts.VersionID,
Internal: minio.AdvancedObjectTaggingOptions{ReplicationProxyRequest: "true"},
})
- errs[i] = err
- if remoteTags != nil {
- existingTags[i] = remoteTags.String()
- }
})
}
wg.Wait()
- // Authorize every contacted target before any mutation. A tag-based deny
- // on one copy must not be hidden by a more permissive copy on another.
- var readErr error
- for i, err := range errs {
- if code := authorize(existingTags[i], err); code != ErrNone {
- return proxy, code
+
+ var firstNotFound, readErr error
+ for _, err := range errs {
+ switch {
+ case err == nil:
+ case isRemoteTaggingNotFound(err):
+ if firstNotFound == nil {
+ firstNotFound = err
+ }
+ default:
+ if code := authorize("", err); code != ErrNone {
+ return nil, proxyResult{}, code
+ }
+ if readErr == nil {
+ readErr = err
+ }
}
- if err != nil && !isRemoteTaggingNotFound(err) && readErr == nil {
- readErr = err
+ }
+ present := make([]taggingProxyTarget, 0, len(targets))
+ for i, tgt := range targets {
+ if errs[i] != nil {
+ continue
}
+ if tgt.tags == nil {
+ tgt.tags, _ = tags.MapToObjectTags(nil)
+ }
+ if code := authorize(tgt.tags.String(), nil); code != ErrNone {
+ return nil, proxyResult{}, code
+ }
+ present = append(present, tgt)
}
if readErr != nil {
- return proxyResult{Err: readErr}, ErrNone
+ return nil, proxyResult{Err: readErr}, ErrNone
+ }
+ if len(present) == 0 {
+ // With no object anywhere, authorize the empty tag set once before
+ // disclosing its absence. No write RPC can help a missing replica.
+ return nil, proxyResult{Err: firstNotFound}, authorize("", firstNotFound)
+ }
+ return present, proxyResult{}, ErrNone
+}
+
+// proxyTaggingToRepTarget proxies tagging requests to remote targets for
+// active-active replicated setups.
+func proxyTaggingToRepTarget(ctx context.Context, bucket, object string, tags *tags.Tags, opts ObjectOptions, proxyTargets *madmin.BucketTargets, authorize func(string, error) APIErrorCode) (proxy proxyResult, authErr APIErrorCode) {
+ if opts.ProxyRequest || (opts.ProxyHeaderSet && !opts.ProxyRequest) {
+ return proxy, ErrNone
+ }
+ targets, proxy, authErr := readTaggingProxyTargets(ctx, bucket, object, opts, proxyTargets, authorize)
+ if authErr != ErrNone || proxy.Err != nil {
+ return proxy, authErr
}
- for i, tgt := range targets {
+ var wg sync.WaitGroup
+ writeErrs := make([]error, len(targets))
+ for i, target := range targets {
wg.Go(func() {
+ tgt := target.client
if tags != nil {
- errs[i] = tgt.PutObjectTagging(ctx, tgt.Bucket, object, tags, minio.PutObjectTaggingOptions{
+ writeErrs[i] = tgt.PutObjectTagging(ctx, tgt.Bucket, object, tags, minio.PutObjectTaggingOptions{
VersionID: opts.VersionID,
Internal: minio.AdvancedObjectTaggingOptions{ReplicationProxyRequest: "true"},
})
} else {
- errs[i] = tgt.RemoveObjectTagging(ctx, tgt.Bucket, object, minio.RemoveObjectTaggingOptions{
+ writeErrs[i] = tgt.RemoveObjectTagging(ctx, tgt.Bucket, object, minio.RemoveObjectTaggingOptions{
VersionID: opts.VersionID,
Internal: minio.AdvancedObjectTaggingOptions{ReplicationProxyRequest: "true"},
})
@@ -2841,9 +2881,8 @@ func proxyTaggingToRepTarget(ctx context.Context, bucket, object string, tags *t
})
}
wg.Wait()
- // Preserve partial success, counting only targets actually contacted.
var lastErr error
- for _, err := range errs {
+ for _, err := range writeErrs {
if err == nil {
return proxy, ErrNone
}
@@ -2853,72 +2892,16 @@ func proxyTaggingToRepTarget(ctx context.Context, bucket, object string, tags *t
}
// proxyGetTaggingToRepTarget proxies get tagging requests to remote targets for
-// active-active replicated setups
-func proxyGetTaggingToRepTarget(ctx context.Context, bucket, object string, opts ObjectOptions, proxyTargets *madmin.BucketTargets) (tgs *tags.Tags, proxy proxyResult) {
- // this option is set when active-active replication is in place between site A -> B,
- // and request hits site B that does not have the object yet.
- if opts.ProxyRequest || (opts.ProxyHeaderSet && !opts.ProxyRequest) { // true only when site B sets MinIOSourceProxyRequest header
- return nil, proxy
+// active-active replicated setups.
+func proxyGetTaggingToRepTarget(ctx context.Context, bucket, object string, opts ObjectOptions, proxyTargets *madmin.BucketTargets, authorize func(string, error) APIErrorCode) (tgs *tags.Tags, proxy proxyResult, authErr APIErrorCode) {
+ if opts.ProxyRequest || (opts.ProxyHeaderSet && !opts.ProxyRequest) {
+ return nil, proxy, ErrNone
}
- var wg sync.WaitGroup
- errs := make([]error, len(proxyTargets.Targets))
- tagSlc := make([]map[string]string, len(proxyTargets.Targets))
- contacted := 0
- for idx, t := range proxyTargets.Targets {
- tgt := globalBucketTargetSys.GetRemoteTargetClient(bucket, t.Arn)
- if tgt == nil || globalBucketTargetSys.isOffline(tgt.EndpointURL()) {
- continue
- }
- // if proxying explicitly disabled on remote target
- if tgt.disableProxy {
- continue
- }
- contacted++
- idx := idx
- wg.Add(1)
- go func(idx int, tgt *TargetClient) {
- defer wg.Done()
- gopts := minio.GetObjectTaggingOptions{
- VersionID: opts.VersionID,
- Internal: minio.AdvancedObjectTaggingOptions{
- ReplicationProxyRequest: "true",
- },
- }
- remoteTags, err := tgt.GetObjectTagging(ctx, tgt.Bucket, object, gopts)
- if err != nil {
- errs[idx] = err
- } else if remoteTags != nil {
- tagSlc[idx] = remoteTags.ToMap()
- }
- }(idx, tgt)
+ targets, proxy, authErr := readTaggingProxyTargets(ctx, bucket, object, opts, proxyTargets, authorize)
+ if authErr != ErrNone || proxy.Err != nil {
+ return nil, proxy, authErr
}
- wg.Wait()
- if contacted == 0 {
- return nil, proxyResult{Err: RemoteTargetConnectionErr{
- Bucket: bucket,
- Err: errors.New("no available remote replication target"),
- }}
- }
- var firstNotFound error
- for idx, err := range errs {
- errCode := minio.ToErrorResponse(err).Code
- if err != nil {
- if errCode != "NoSuchKey" && errCode != "NoSuchVersion" {
- return nil, proxyResult{Err: err}
- }
- if firstNotFound == nil {
- firstNotFound = err
- }
- continue
- }
- if tagSlc[idx] != nil && tgs == nil {
- tgs, _ = tags.MapToObjectTags(tagSlc[idx])
- }
- }
- if tgs == nil {
- proxy.Err = firstNotFound
- }
- return tgs, proxy
+ return targets[0].tags, proxy, ErrNone
}
func scheduleReplicationDelete(ctx context.Context, dv DeletedObjectReplicationInfo, o ObjectLayer) {
diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go
index 13c122f62..e848a5ce4 100644
--- a/cmd/object-handlers.go
+++ b/cmd/object-handlers.go
@@ -3726,9 +3726,22 @@ func (api objectAPIHandlers) GetObjectTaggingHandler(w http.ResponseWriter, r *h
if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
proxytgts := getProxyTargets(ctx, bucket, object, opts)
if !proxytgts.Empty() {
+ if s3Error := authorizeRequestWithPotentialTags(ctx, r, policy.GetObjectTaggingAction); s3Error != ErrNone {
+ writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+ return
+ }
globalReplicationStats.Load().incProxy(bucket, getObjectTaggingAPI, false)
// proxy to replication target if site replication is in place.
- tags, gerr := proxyGetTaggingToRepTarget(ctx, bucket, object, opts, proxytgts)
+ tags, gerr, s3Error := proxyGetTaggingToRepTarget(ctx, bucket, object, opts, proxytgts, func(existingTags string, readErr error) APIErrorCode {
+ if readErr != nil && !isRemoteTaggingNotFound(readErr) {
+ return authorizeRequestWithUnreadTags(ctx, r, policy.GetObjectTaggingAction, readErr)
+ }
+ return authorizeRequestWithExistingTags(ctx, r, policy.GetObjectTaggingAction, existingTags)
+ })
+ if s3Error != ErrNone {
+ writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+ return
+ }
if gerr.Err != nil || tags == nil {
globalReplicationStats.Load().incProxy(bucket, getObjectTaggingAPI, true)
writeErrorResponse(ctx, w, toAPIError(ctx, gerr.Err), r.URL)
@@ -3737,10 +3750,18 @@ func (api objectAPIHandlers) GetObjectTaggingHandler(w http.ResponseWriter, r *h
ot = tags
w.Header()[xhttp.MinIOTaggingProxied] = []string{"true"} // indicate that the request was proxied.
} else {
+ if s3Error := authorizeRequestWithExistingTags(ctx, r, policy.GetObjectTaggingAction, ""); s3Error != ErrNone {
+ writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+ return
+ }
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
} else {
+ if s3Error := authorizeRequestWithUnreadTags(ctx, r, policy.GetObjectTaggingAction, err); s3Error != ErrNone {
+ writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+ return
+ }
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
@@ -3841,6 +3862,10 @@ func (api objectAPIHandlers) PutObjectTaggingHandler(w http.ResponseWriter, r *h
if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
proxytgts := getProxyTargets(ctx, bucket, object, opts)
if !proxytgts.Empty() {
+ if s3Error := authorizeRequestWithPotentialTags(ctx, r, policy.PutObjectTaggingAction); s3Error != ErrNone {
+ writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+ return
+ }
perr, s3Error := proxyTaggingToRepTarget(ctx, bucket, object, tags, opts, proxytgts, func(existingTags string, readErr error) APIErrorCode {
if readErr != nil && !isRemoteTaggingNotFound(readErr) {
return authorizeRequestWithUnreadTags(ctx, r, policy.PutObjectTaggingAction, readErr)
@@ -3969,6 +3994,10 @@ func (api objectAPIHandlers) DeleteObjectTaggingHandler(w http.ResponseWriter, r
if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
proxytgts := getProxyTargets(ctx, bucket, object, opts)
if !proxytgts.Empty() {
+ if s3Error := authorizeRequestWithPotentialTags(ctx, r, policy.DeleteObjectTaggingAction); s3Error != ErrNone {
+ writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+ return
+ }
perr, s3Error := proxyTaggingToRepTarget(ctx, bucket, object, nil, opts, proxytgts, func(existingTags string, readErr error) APIErrorCode {
if readErr != nil && !isRemoteTaggingNotFound(readErr) {
return authorizeRequestWithUnreadTags(ctx, r, policy.DeleteObjectTaggingAction, readErr)
@@ -4001,7 +4030,13 @@ func (api objectAPIHandlers) DeleteObjectTaggingHandler(w http.ResponseWriter, r
return
}
}
- if s3Error := authorizeRequestWithExistingTags(ctx, r, policy.DeleteObjectTaggingAction, ""); s3Error != ErrNone {
+ var s3Error APIErrorCode
+ if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
+ s3Error = authorizeRequestWithExistingTags(ctx, r, policy.DeleteObjectTaggingAction, "")
+ } else {
+ s3Error = authorizeRequestWithUnreadTags(ctx, r, policy.DeleteObjectTaggingAction, err)
+ }
+ if s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
diff --git a/cmd/tagging-proxy-authz_test.go b/cmd/tagging-proxy-authz_test.go
index 0e0f0eb77..e50fc8343 100644
--- a/cmd/tagging-proxy-authz_test.go
+++ b/cmd/tagging-proxy-authz_test.go
@@ -33,11 +33,18 @@ import (
func TestAPITaggingProxyAuthorization(t *testing.T) {
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: func(obj ObjectLayer, backend, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
conditional := newDeleteAuthzPolicyUser(t, backend, bucket, fmt.Sprintf(`[
- {"Effect":"Allow","Action":["s3:PutObjectTagging","s3:DeleteObjectTagging"],"Resource":["arn:aws:s3:::%s/*"]},
- {"Effect":"Deny","Action":["s3:PutObjectTagging","s3:DeleteObjectTagging"],"Resource":["arn:aws:s3:::%s/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/secret":"true"}}}
+ {"Effect":"Allow","Action":["s3:GetObjectTagging","s3:PutObjectTagging","s3:DeleteObjectTagging"],"Resource":["arn:aws:s3:::%s/*"]},
+ {"Effect":"Deny","Action":["s3:GetObjectTagging","s3:PutObjectTagging","s3:DeleteObjectTagging"],"Resource":["arn:aws:s3:::%s/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/secret":"true"}}}
]`, bucket, bucket))
+ tagAllow := newDeleteAuthzPolicyUser(t, backend, bucket, fmt.Sprintf(`[
+ {"Effect":"Allow","Action":["s3:GetObjectTagging","s3:PutObjectTagging","s3:DeleteObjectTagging"],"Resource":["arn:aws:s3:::%s/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/secret":"false"}}}
+ ]`, bucket))
+ denied := newDeleteAuthzPolicyUser(t, backend, bucket, fmt.Sprintf(`[
+ {"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::%s/*"]}
+ ]`, bucket))
+ zero := iamAdminUser(t.Context(), t, globalIAMSys, "tagging-no-policy-"+mustGetUUID(), "")
versionID := mustGetUUID()
- for _, method := range []string{http.MethodPut, http.MethodDelete} {
+ for _, method := range []string{http.MethodGet, http.MethodPut, http.MethodDelete} {
for _, tc := range []struct {
name string
tags []string
@@ -45,17 +52,32 @@ func TestAPITaggingProxyAuthorization(t *testing.T) {
readStatus, writeStatus, wantStatus int
wantWrites int32
disabledFirst bool
+ firstReadStatus int
+ noReads bool
}{
{name: "deny-second-target", tags: []string{"false", "true"}, cred: conditional, wantStatus: http.StatusForbidden},
{name: "deny-first-target", tags: []string{"true", "false"}, cred: conditional, wantStatus: http.StatusForbidden},
{name: "all-authorized", tags: []string{"false", "false"}, cred: conditional, wantWrites: 2},
- {name: "anonymous", tags: []string{"false"}, wantStatus: http.StatusForbidden},
- {name: "invalid-credentials", tags: []string{"false"}, cred: auth.Credentials{AccessKey: root.AccessKey, SecretKey: "wrong-secret-key"}, wantStatus: http.StatusForbidden},
+ {name: "anonymous", tags: []string{"false"}, wantStatus: http.StatusForbidden, noReads: true},
+ {name: "invalid-credentials", tags: []string{"false"}, cred: auth.Credentials{AccessKey: root.AccessKey, SecretKey: "wrong-secret-key"}, wantStatus: http.StatusForbidden, noReads: true},
+ {name: "zero-permissions", tags: []string{"false"}, cred: zero, wantStatus: http.StatusForbidden, noReads: true},
+ {name: "zero-permissions-missing", tags: []string{"false"}, cred: zero, readStatus: http.StatusNotFound, wantStatus: http.StatusForbidden, noReads: true},
+ {name: "unconditional-deny", tags: []string{"false"}, cred: denied, wantStatus: http.StatusForbidden, noReads: true},
{name: "unreadable", tags: []string{"false"}, cred: root, readStatus: http.StatusServiceUnavailable, wantStatus: http.StatusServiceUnavailable},
- {name: "missing-version", tags: []string{"false"}, cred: root, readStatus: http.StatusNotFound, writeStatus: http.StatusNotFound, wantStatus: http.StatusNotFound, wantWrites: 1},
+ {name: "missing-version", tags: []string{"false", "false"}, cred: root, readStatus: http.StatusNotFound, wantStatus: http.StatusNotFound},
+ {name: "tag-allow-missing-version", tags: []string{"false"}, cred: tagAllow, readStatus: http.StatusNotFound, wantStatus: http.StatusForbidden},
+ {name: "tag-allow-lagging-first", tags: []string{"false", "false"}, cred: tagAllow, firstReadStatus: http.StatusNotFound, wantWrites: 1},
+ {name: "tag-allow-lagging-second", tags: []string{"false", "false"}, cred: tagAllow, firstReadStatus: http.StatusOK, readStatus: http.StatusNotFound, wantWrites: 1},
+ {name: "tag-allow-present-untagged", tags: []string{"false", ""}, cred: tagAllow, wantStatus: http.StatusForbidden},
+ {name: "tag-allow-unreadable-first", tags: []string{"false", "false"}, cred: tagAllow, firstReadStatus: http.StatusServiceUnavailable, wantStatus: http.StatusServiceUnavailable},
+ {name: "tag-allow-unreadable-second", tags: []string{"false", "false"}, cred: tagAllow, firstReadStatus: http.StatusOK, readStatus: http.StatusServiceUnavailable, wantStatus: http.StatusServiceUnavailable},
{name: "disabled-target", tags: []string{"true", "false"}, cred: conditional, disabledFirst: true, wantWrites: 1},
+ {name: "no-available-target", tags: []string{"false"}, cred: tagAllow, disabledFirst: true, wantStatus: http.StatusServiceUnavailable, noReads: true},
{name: "skipped-target-cannot-mask-failure", tags: []string{"false", "false"}, cred: root, disabledFirst: true, writeStatus: http.StatusServiceUnavailable, wantStatus: http.StatusServiceUnavailable, wantWrites: 1},
} {
+ if method == http.MethodGet && tc.writeStatus != 0 {
+ continue // This case only exercises a remote mutation failure.
+ }
t.Run(backend+"/"+method+"/"+tc.name, func(t *testing.T) {
var reads, writes atomic.Int32
globalBucketTargetSys.Lock()
@@ -63,6 +85,10 @@ func TestAPITaggingProxyAuthorization(t *testing.T) {
globalBucketTargetSys.Unlock()
cfg := replication.Config{}
for i, value := range tc.tags {
+ readStatus := tc.readStatus
+ if i == 0 && tc.firstReadStatus != 0 {
+ readStatus = tc.firstReadStatus
+ }
target := newILMTestTarget(t)
registerILMTestTarget(t, obj, bucket, target, func(w http.ResponseWriter, r *http.Request) {
if r.URL.Query().Get("versionId") != versionID || !r.URL.Query().Has("tagging") || r.Header.Get(xhttp.MinIOSourceProxyRequest) != "true" {
@@ -71,12 +97,15 @@ func TestAPITaggingProxyAuthorization(t *testing.T) {
status := tc.writeStatus
if r.Method == http.MethodGet {
reads.Add(1)
- status = tc.readStatus
+ status = readStatus
} else {
writes.Add(1)
+ if readStatus == http.StatusNotFound {
+ t.Error("tagging write reached a missing replica")
+ }
}
w.Header().Set("Content-Type", "application/xml")
- if status != 0 {
+ if status != 0 && status != http.StatusOK {
w.WriteHeader(status)
code := "ServiceUnavailable"
if status == http.StatusNotFound {
@@ -87,7 +116,11 @@ func TestAPITaggingProxyAuthorization(t *testing.T) {
}
switch r.Method {
case http.MethodGet:
- fmt.Fprintf(w, "secret%s", value)
+ if value == "" {
+ fmt.Fprint(w, "")
+ } else {
+ fmt.Fprintf(w, "secret%s", value)
+ }
case http.MethodDelete:
w.WriteHeader(http.StatusNoContent)
}
@@ -130,11 +163,15 @@ func TestAPITaggingProxyAuthorization(t *testing.T) {
wantStatus = http.StatusNoContent
}
}
- if rec.Code != wantStatus || writes.Load() != tc.wantWrites {
- t.Errorf("status=%d writes=%d, want status=%d writes=%d: %s", rec.Code, writes.Load(), wantStatus, tc.wantWrites, rec.Body.String())
+ wantWrites := tc.wantWrites
+ if method == http.MethodGet {
+ wantWrites = 0
}
- if tc.name == "invalid-credentials" && reads.Load() != 0 {
- t.Errorf("invalid credentials reached %d remote reads", reads.Load())
+ if rec.Code != wantStatus || writes.Load() != wantWrites {
+ t.Errorf("status=%d writes=%d, want status=%d writes=%d: %s", rec.Code, writes.Load(), wantStatus, wantWrites, rec.Body.String())
+ }
+ if tc.noReads && reads.Load() != 0 {
+ t.Errorf("unauthorized request reached %d remote reads", reads.Load())
}
})
}