Merge pull request #2342 from versity/sis/delete-objects-deny-bypass

fix: authorize every key in a DeleteObjects batch after a bucket-policy deny

auth/access-control.go

@@ -283,15 +283,25 @@ func objectsAccessErrors(ctx context.Context, be backend.Backend, opts AccessOpt
 	errs := make([]error, len(keys))
 
 	// An explicit deny from the bucket policy wins outright, whatever the
-	// IAM backend is, so a request carrying one needs no identity policy at
-	// all — which also saves the standalone IAM service round trip. Only the
-	// first denied key is recorded: the request fails there regardless of
-	// what the rest would have evaluated to.
+	// IAM backend is and whatever an identity policy would have said, so
+	// every denied key is settled here and never revisited below. Each key
+	// is settled on its own: this is a partial-success API, so a deny on one
+	// key says nothing about the next one, which still has to be evaluated
+	// on its own merits.
+	allDenied := true
 	for i, rd := range resourceDecisions {
 		if rd.Decision == policyDecisionDeny {
 			errs[i] = s3err.GetExplicitDenyAccessErr(opts.Acc.Access, string(rd.Action), objectPolicyArn(opts.Bucket, keys[i], be.NormalizeObjectKey), "a resource-based policy")
-			return errs, nil
+			continue
 		}
+		allDenied = false
+	}
+	// Nothing is left to decide, so skip the identity policy entirely —
+	// which also saves the standalone IAM service round trip. That shortcut
+	// only holds when the bucket policy denied every key; a single
+	// undecided key still needs the identity policy consulted for it.
+	if allDenied {
+		return errs, nil
 	}
 
 	pe, hasPolicyEvaluator := opts.Iam.(PolicyEvaluator)
@@ -300,6 +310,11 @@ func objectsAccessErrors(ctx context.Context, be backend.Backend, opts AccessOpt
 		// today's exact behavior and generic message, unconditionally, for
 		// every internal/LDAP/Vault/IPA/S3-IAM deployment.
 		for i, rd := range resourceDecisions {
+			if errs[i] != nil {
+				// Explicitly denied above — keep that specific message
+				// rather than flattening it to the generic one.
+				continue
+			}
 			if rd.Decision != policyDecisionAllow {
 				errs[i] = s3err.GetAPIError(s3err.ErrAccessDenied)
 			}
@@ -318,6 +333,13 @@ func objectsAccessErrors(ctx context.Context, be backend.Backend, opts AccessOpt
 	}
 
 	for i := range keys {
+		if errs[i] != nil {
+			// Explicitly denied by the bucket policy. An explicit deny is
+			// final, so no identity-policy result can clear it, and the
+			// resource-based message is the one AWS reports for it.
+			continue
+		}
+
 		resourceArn := objectPolicyArn(opts.Bucket, keys[i], be.NormalizeObjectKey)
 
 		if identity.Decisions[i].Decision == policyDecisionDeny {

auth/access-control_test.go

@@ -779,4 +779,142 @@ func TestVerifyObjectsAccess_VersionedDeleteNeedsSeparatePermission(t *testing.T
 	}
 }
 
+// bucketPolicyNoLockBackend serves a fixed bucket policy and answers "no
+// lock configuration", so VerifyObjectsAccess' lock check is a no-op and
+// only the policy half of the per-object result is under test.
+type bucketPolicyNoLockBackend struct {
+	backend.BackendUnsupported
+	policy []byte
+}
+
+func (b bucketPolicyNoLockBackend) GetBucketPolicy(_ context.Context, _ string) ([]byte, error) {
+	return b.policy, nil
+}
+
+func (b bucketPolicyNoLockBackend) GetObjectLockConfiguration(_ context.Context, _ string) ([]byte, error) {
+	return nil, s3err.GetAPIError(s3err.ErrObjectLockConfigurationNotFound)
+}
+
+// denyProtectedPrefixBackend is a bucket policy denying s3:DeleteObject on
+// one prefix and saying nothing about anything else, so a batch can mix
+// explicitly denied keys with keys the bucket policy leaves undecided.
+func denyProtectedPrefixBackend() bucketPolicyNoLockBackend {
+	return bucketPolicyNoLockBackend{policy: []byte(`{
+		"Statement": [{
+			"Effect": "Deny",
+			"Principal": "testuser",
+			"Action": "s3:DeleteObject",
+			"Resource": "arn:aws:s3:::bucket/protected/*"
+		}]
+	}`)}
+}
+
+func deleteObjectIdentifiers(keys ...string) []types.ObjectIdentifier {
+	objects := make([]types.ObjectIdentifier, len(keys))
+	for i, key := range keys {
+		objects[i] = types.ObjectIdentifier{Key: strPtr(key)}
+	}
+	return objects
+}
+
+// TestVerifyObjectsAccess_ResourceDenyDoesNotAuthorizeLaterKeys is the
+// authorization-bypass regression: a bucket-policy Deny used to end the
+// whole batch's evaluation at the first denied key, leaving every later key
+// with a nil result — which VerifyObjectsAccess' caller reads as
+// "authorized" and sends straight to the backend. Every key must be settled
+// on its own instead: the denied one explicitly, the rest by the identity
+// policy, which here allows neither.
+func TestVerifyObjectsAccess_ResourceDenyDoesNotAuthorizeLaterKeys(t *testing.T) {
+	pe := newMockPolicyEvaluator(policyDecisionNoMatch)
+	pe.principalArn = "arn:aws:iam::000000000000:user/testuser"
+
+	errs, err := VerifyObjectsAccess(testFiberCtx(t), denyProtectedPrefixBackend(), AccessOptions{
+		Acc:           Account{Access: "testuser", Role: RoleUser},
+		Bucket:        "bucket",
+		AclPermission: PermissionWrite,
+		Iam:           pe,
+	}, deleteObjectIdentifiers("protected/x", "secret/y"), BypassNone)
+
+	assert.NoError(t, err)
+	if assert.Len(t, errs, 2) {
+		denied := requireAccessDeniedAPIError(t, errs[0])
+		assert.Contains(t, denied.Description, "with an explicit deny in a resource-based policy")
+
+		later := requireAccessDeniedAPIError(t, errs[1])
+		assert.Contains(t, later.Description, "because no identity-based policy allows the s3:DeleteObject action")
+	}
+	assert.Len(t, pe.calls, 1, "a key the bucket policy left undecided still needs the identity policy consulted for it")
+}
+
+// TestVerifyObjectsAccess_ResourceDenyKeptOverIdentityAllow confirms an
+// explicit deny still wins per key once every key is evaluated: the denied
+// key keeps its resource-based denial even though the identity policy
+// allows it, while the key the bucket policy said nothing about is
+// authorized by that same identity Allow.
+func TestVerifyObjectsAccess_ResourceDenyKeptOverIdentityAllow(t *testing.T) {
+	pe := newMockPolicyEvaluator(policyDecisionAllow)
+
+	errs, err := VerifyObjectsAccess(testFiberCtx(t), denyProtectedPrefixBackend(), AccessOptions{
+		Acc:           Account{Access: "testuser", Role: RoleUser},
+		Bucket:        "bucket",
+		AclPermission: PermissionWrite,
+		Iam:           pe,
+	}, deleteObjectIdentifiers("protected/x", "allowed/y"), BypassNone)
+
+	assert.NoError(t, err)
+	if assert.Len(t, errs, 2) {
+		denied := requireAccessDeniedAPIError(t, errs[0])
+		assert.Contains(t, denied.Description, "with an explicit deny in a resource-based policy")
+		assert.NoError(t, errs[1], "the identity policy's Allow stands for the key the bucket policy didn't deny")
+	}
+}
+
+// TestVerifyObjectsAccess_AllKeysResourceDeniedSkipsIdentityPolicy covers
+// the round trip the short-circuit was there to save: it is still skipped,
+// but only when the bucket policy denied every key in the batch, since then
+// no identity-policy answer could change any result.
+func TestVerifyObjectsAccess_AllKeysResourceDeniedSkipsIdentityPolicy(t *testing.T) {
+	pe := newMockPolicyEvaluator(policyDecisionAllow)
+
+	errs, err := VerifyObjectsAccess(testFiberCtx(t), denyProtectedPrefixBackend(), AccessOptions{
+		Acc:           Account{Access: "testuser", Role: RoleUser},
+		Bucket:        "bucket",
+		AclPermission: PermissionWrite,
+		Iam:           pe,
+	}, deleteObjectIdentifiers("protected/x", "protected/y"), BypassNone)
+
+	assert.NoError(t, err)
+	if assert.Len(t, errs, 2) {
+		for i, e := range errs {
+			denied := requireAccessDeniedAPIError(t, e)
+			assert.Containsf(t, denied.Description, "with an explicit deny in a resource-based policy", "key %d", i)
+		}
+	}
+	assert.Empty(t, pe.calls, "with every key already explicitly denied there is nothing left for the identity policy to decide")
+}
+
+// TestVerifyObjectsAccess_ResourceDenyNoPolicyEvaluator covers the same
+// bypass for the IAM backends with no identity-policy layer at all
+// (internal, LDAP, Vault, IPA): the denied key keeps its specific message
+// and every other key falls to the generic AccessDenied those backends have
+// always returned — none of them silently authorized.
+func TestVerifyObjectsAccess_ResourceDenyNoPolicyEvaluator(t *testing.T) {
+	errs, err := VerifyObjectsAccess(testFiberCtx(t), denyProtectedPrefixBackend(), AccessOptions{
+		Acc:           Account{Access: "testuser", Role: RoleUser},
+		Bucket:        "bucket",
+		AclPermission: PermissionWrite,
+		Iam:           NewIAMServiceSingle(Account{}),
+	}, deleteObjectIdentifiers("protected/x", "secret/y"), BypassNone)
+
+	assert.NoError(t, err)
+	if assert.Len(t, errs, 2) {
+		denied := requireAccessDeniedAPIError(t, errs[0])
+		assert.Contains(t, denied.Description, "with an explicit deny in a resource-based policy")
+
+		later := requireAccessDeniedAPIError(t, errs[1])
+		assert.Equal(t, s3err.GetAPIError(s3err.ErrAccessDenied).Description, later.Description,
+			"backends with no identity-policy layer keep their generic message")
+	}
+}
+
 func strPtr(s string) *string { return &s }

tests/integration/DeleteObjects.go

@@ -210,10 +210,7 @@ func DeleteObjects_iam_mixed_denials_and_success(s *S3Conf) error {
 		if err := checkDeletedKeysInOrder(out.Deleted, []string{"allowed/one", "allowed/two"}); err != nil {
 			return err
 		}
-		return checkDeleteObjectsErrsInOrder(out.Errors, []struct {
-			key string
-			err s3err.S3Error
-		}{
+		return checkDeleteObjectsErrsInOrder(out.Errors, []keyDenial{
 			{"denied/one", wantImplicitDeny(user.arn, actS3DeleteObject, objectArn(bucket, "denied/one"))},
 			{"locked/one", s3err.GetAPIError(s3err.ErrObjectLocked)},
 		})
@@ -258,10 +255,7 @@ func DeleteObjects_iam_all_access_denied(s *S3Conf) error {
 		if len(out.Deleted) != 0 {
 			return fmt.Errorf("expected nothing deleted, got %+v", out.Deleted)
 		}
-		return checkDeleteObjectsErrsInOrder(out.Errors, []struct {
-			key string
-			err s3err.S3Error
-		}{
+		return checkDeleteObjectsErrsInOrder(out.Errors, []keyDenial{
 			{"one", wantImplicitDeny(user.arn, actS3DeleteObject, objectArn(bucket, "one"))},
 			{"two", wantImplicitDeny(user.arn, actS3DeleteObject, objectArn(bucket, "two"))},
 			{"three", wantImplicitDeny(user.arn, actS3DeleteObject, objectArn(bucket, "three"))},
@@ -311,10 +305,7 @@ func DeleteObjects_iam_all_locked(s *S3Conf) error {
 		if len(out.Deleted) != 0 {
 			return fmt.Errorf("expected nothing deleted, got %+v", out.Deleted)
 		}
-		if err := checkDeleteObjectsErrsInOrder(out.Errors, []struct {
-			key string
-			err s3err.S3Error
-		}{
+		if err := checkDeleteObjectsErrsInOrder(out.Errors, []keyDenial{
 			{"locked/one", s3err.GetAPIError(s3err.ErrObjectLocked)},
 			{"locked/two", s3err.GetAPIError(s3err.ErrObjectLocked)},
 		}); err != nil {
@@ -337,10 +328,7 @@ func DeleteObjects_iam_all_locked(s *S3Conf) error {
 		if len(out.Deleted) != 0 {
 			return fmt.Errorf("expected nothing deleted, got %+v", out.Deleted)
 		}
-		if err := checkDeleteObjectsErrsInOrder(out.Errors, []struct {
-			key string
-			err s3err.S3Error
-		}{
+		if err := checkDeleteObjectsErrsInOrder(out.Errors, []keyDenial{
 			{"locked/one", wantImplicitDeny(user.arn, actS3BypassGovernance, objectArn(bucket, "locked/one"))},
 			{"locked/two", wantImplicitDeny(user.arn, actS3BypassGovernance, objectArn(bucket, "locked/two"))},
 		}); err != nil {

tests/integration/group-tests.go

@@ -1721,6 +1721,8 @@ func TestS3IAMAccessControl(ts *TestState) {
 	ts.Run(S3IAMAccessControl_compliance_mode_not_bypassable)
 	ts.Run(S3IAMAccessControl_delete_objects_authorizes_each_key)
 	ts.Run(S3IAMAccessControl_delete_objects_version_needs_separate_permission)
+	ts.Run(S3IAMAccessControl_delete_objects_bucket_policy_deny_per_key)
+	ts.Run(S3IAMAccessControl_delete_objects_deny_across_versioned_split)
 	ts.Run(S3IAMAccessControl_governance_bypass_delete_objects)
 	ts.Run(DeleteObjects_iam_mixed_denials_and_success)
 	ts.Run(DeleteObjects_iam_all_access_denied)
@@ -2139,6 +2141,8 @@ func GetIntTests() IntTests {
 		"S3IAMAccessControl_retention_extension_needs_no_bypass":                           S3IAMAccessControl_retention_extension_needs_no_bypass,
 		"S3IAMAccessControl_delete_objects_authorizes_each_key":                            S3IAMAccessControl_delete_objects_authorizes_each_key,
 		"S3IAMAccessControl_delete_objects_version_needs_separate_permission":              S3IAMAccessControl_delete_objects_version_needs_separate_permission,
+		"S3IAMAccessControl_delete_objects_bucket_policy_deny_per_key":                     S3IAMAccessControl_delete_objects_bucket_policy_deny_per_key,
+		"S3IAMAccessControl_delete_objects_deny_across_versioned_split":                    S3IAMAccessControl_delete_objects_deny_across_versioned_split,
 		"S3IAMAccessControl_no_policy_denies":                                              S3IAMAccessControl_no_policy_denies,
 		"S3IAMAccessControl_root_bypasses_policies":                                        S3IAMAccessControl_root_bypasses_policies,
 		"S3IAMAccessControl_identity_policy_allows_without_bucket_policy":                  S3IAMAccessControl_identity_policy_allows_without_bucket_policy,

tests/integration/s3_iam_access_control.go

@@ -984,8 +984,7 @@ func S3IAMAccessControl_compliance_mode_not_bypassable(s *S3Conf) error {
 // S3IAMAccessControl_delete_objects_authorizes_each_key verifies the batch
 // DeleteObjects path authorizes s3:DeleteObject against each object's own
 // ARN, the way real AWS does — a policy naming only "bucket/*" is
-// sufficient — and that it supports partial success: verified live against
-// real AWS (niksis02, account 792168558830), a key outside the granted
+// sufficient — and that it supports partial success: a key outside the granted
 // prefix denies only that key, reported in the response's Errors list, while
 // every other key in the same batch is still deleted and reported in
 // Deleted. Both lists preserve the order the keys were requested in.
@@ -1046,18 +1045,10 @@ func S3IAMAccessControl_delete_objects_authorizes_each_key(s *S3Conf) error {
 // S3IAMAccessControl_delete_objects_version_needs_separate_permission
 // verifies that naming a VersionId in a DeleteObjects entry is authorized
 // against s3:DeleteObjectVersion, a distinct permission from the
-// s3:DeleteObject a keyed (unversioned) delete needs — verified live against
-// real AWS (niksis02, account 792168558830): a policy granting only
+// s3:DeleteObject a keyed (unversioned) delete needs - a policy granting only
 // s3:DeleteObject denies the versioned deletes in a batch while its keyed
 // deletes in the same batch still succeed, each independently, matching the
 // single-object DELETE path's existing behavior for the same distinction.
-//
-// The denial happens at authorization, before the backend ever resolves the
-// named version, so this doesn't need a real object version (and the
-// gateway this test group runs against has no --versioning-dir configured
-// to produce one): an arbitrary VersionId is enough to exercise the
-// s3:DeleteObjectVersion check and prove the batch still partially
-// succeeds.
 func S3IAMAccessControl_delete_objects_version_needs_separate_permission(s *S3Conf) error {
 	testName := "S3IAMAccessControl_delete_objects_version_needs_separate_permission"
 	return s3IAMActionHandler(s, testName, func(root *iam.Client, bucket string) error {
@@ -1108,6 +1099,181 @@ func S3IAMAccessControl_delete_objects_version_needs_separate_permission(s *S3Co
 	})
 }
 
+// S3IAMAccessControl_delete_objects_bucket_policy_deny_per_key verifies a
+// bucket-policy Deny settles only the key it names. DeleteObjects is a
+// partial-success API, so every key is authorized on its own: a Deny on one
+// key is reported against that key and leaves every other key to be
+// evaluated on its own merits — an explicitly denied key never lets the
+// keys after it through unauthorized.
+func S3IAMAccessControl_delete_objects_bucket_policy_deny_per_key(s *S3Conf) error {
+	testName := "S3IAMAccessControl_delete_objects_bucket_policy_deny_per_key"
+	return s3IAMActionHandler(s, testName, func(root *iam.Client, bucket string) error {
+		user, cleanup, err := newS3IAMUser(root, s, map[string]string{
+			"p": policyDoc(accessStatement{
+				Effect: "Allow", Action: actS3DeleteObject,
+				Resource: objectArn(bucket, "allowed/*"),
+			}),
+		})
+		if err != nil {
+			return err
+		}
+		defer cleanup()
+
+		// Two Deny statements, so a batch can carry two explicitly denied
+		// keys — the second one proves the first didn't end the evaluation.
+		if err := putBucketPolicyDoc(s, bucket,
+			bucketStatement{
+				Effect: "Deny", Principal: user.conf.awsID,
+				Action: actS3DeleteObject, Resource: objectArn(bucket, "protected/*"),
+			},
+			bucketStatement{
+				Effect: "Deny", Principal: user.conf.awsID,
+				Action: actS3DeleteObject, Resource: objectArn(bucket, "vault/*"),
+			},
+		); err != nil {
+			return err
+		}
+
+		// The three denial shapes this group's keys produce: "protected/"
+		// and "vault/" are explicitly denied by the bucket policy, "secret/"
+		// is denied for want of any grant, and "allowed/" is the only prefix
+		// the identity policy permits.
+		resourceDeny := func(key string) keyDenial {
+			return keyDenial{key, wantExplicitResourceDeny(user.conf.awsID, actS3DeleteObject, objectArn(bucket, key))}
+		}
+		implicitDeny := func(key string) keyDenial {
+			return keyDenial{key, wantImplicitDeny(user.arn, actS3DeleteObject, objectArn(bucket, key))}
+		}
+
+		for _, tc := range []struct {
+			name        string
+			keys        []string
+			wantErrs    []keyDenial
+			wantDeleted []string
+		}{
+			{
+				name:     "a bucket-policy deny and an unauthorized key",
+				keys:     []string{"protected/x", "secret/z"},
+				wantErrs: []keyDenial{resourceDeny("protected/x"), implicitDeny("secret/z")},
+			},
+			{
+				name:        "a bucket-policy deny and an authorized key",
+				keys:        []string{"protected/x", "allowed/w"},
+				wantErrs:    []keyDenial{resourceDeny("protected/x")},
+				wantDeleted: []string{"allowed/w"},
+			},
+			{
+				name:     "two bucket-policy denies",
+				keys:     []string{"protected/x", "vault/y"},
+				wantErrs: []keyDenial{resourceDeny("protected/x"), resourceDeny("vault/y")},
+			},
+			// The same batches with the keys reversed: the outcome depends
+			// on each key, never on where in the batch a denial first
+			// appeared.
+			{
+				name:     "an unauthorized key before a bucket-policy deny",
+				keys:     []string{"secret/z", "protected/x"},
+				wantErrs: []keyDenial{implicitDeny("secret/z"), resourceDeny("protected/x")},
+			},
+			{
+				name:        "an authorized key before a bucket-policy deny",
+				keys:        []string{"allowed/w", "protected/x"},
+				wantErrs:    []keyDenial{resourceDeny("protected/x")},
+				wantDeleted: []string{"allowed/w"},
+			},
+			{
+				name:     "two bucket-policy denies, reversed",
+				keys:     []string{"vault/y", "protected/x"},
+				wantErrs: []keyDenial{resourceDeny("vault/y"), resourceDeny("protected/x")},
+			},
+		} {
+			if err := func() error {
+				// Root rewrites every key before each case: a case that
+				// deletes one must not change what the next one sees.
+				if _, err := putObjects(s.GetClient(), []string{"protected/x", "vault/y", "secret/z", "allowed/w"}, bucket); err != nil {
+					return err
+				}
+
+				out, err := deleteObjectsBatch(user.client, bucket, tc.keys...)
+				if err != nil {
+					return fmt.Errorf("expected DeleteObjects to succeed with per-object denials, not fail outright: %w", err)
+				}
+				if err := checkDeletedKeysInOrder(out.Deleted, tc.wantDeleted); err != nil {
+					return err
+				}
+				return checkDeleteObjectsErrsInOrder(out.Errors, tc.wantErrs)
+			}(); err != nil {
+				return fmt.Errorf("%s: %w", tc.name, err)
+			}
+		}
+		return nil
+	})
+}
+
+// S3IAMAccessControl_delete_objects_deny_across_versioned_split verifies a
+// denial in one half of a batch doesn't authorize the other half. A batch is
+// split by action — keyed entries are authorized against s3:DeleteObject,
+// entries naming a VersionId against s3:DeleteObjectVersion — and each half
+// is evaluated separately, so a denial has to settle its own key in its own
+// half and nothing else: not the keys after it, and not the keys in the
+// other half.
+func S3IAMAccessControl_delete_objects_deny_across_versioned_split(s *S3Conf) error {
+	testName := "S3IAMAccessControl_delete_objects_deny_across_versioned_split"
+	return s3IAMActionHandler(s, testName, func(root *iam.Client, bucket string) error {
+		if _, err := putObjects(s.GetClient(), []string{"protected/x", "secret/z", "allowed/w"}, bucket); err != nil {
+			return err
+		}
+
+		user, cleanup, err := newS3IAMUser(root, s, map[string]string{
+			"p": policyDoc(accessStatement{
+				Effect: "Allow", Action: actS3DeleteObject,
+				Resource: objectArn(bucket, "allowed/*"),
+			}),
+		})
+		if err != nil {
+			return err
+		}
+		defer cleanup()
+
+		if err := putBucketPolicyDoc(s, bucket, bucketStatement{
+			Effect: "Deny", Principal: user.conf.awsID,
+			Action:   []string{actS3DeleteObject, actS3DeleteObjectVersion},
+			Resource: objectArn(bucket, "protected/*"),
+		}); err != nil {
+			return err
+		}
+
+		// Both halves interleaved, each led by its explicitly denied key, so
+		// that a leak in either half shows up as a key deleted or missing
+		// from Errors.
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := user.client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
+			Bucket: &bucket,
+			Delete: &types.Delete{Objects: []types.ObjectIdentifier{
+				{Key: aws.String("protected/x")},
+				{Key: aws.String("protected/v"), VersionId: aws.String("some-version-id")},
+				{Key: aws.String("secret/z")},
+				{Key: aws.String("other/v"), VersionId: aws.String("some-version-id")},
+				{Key: aws.String("allowed/w")},
+			}},
+		})
+		cancel()
+		if err != nil {
+			return fmt.Errorf("expected DeleteObjects to succeed with per-object denials, not fail outright: %w", err)
+		}
+
+		if err := checkDeletedKeysInOrder(out.Deleted, []string{"allowed/w"}); err != nil {
+			return err
+		}
+		return checkDeleteObjectsErrsInOrder(out.Errors, []keyDenial{
+			{"protected/x", wantExplicitResourceDeny(user.conf.awsID, actS3DeleteObject, objectArn(bucket, "protected/x"))},
+			{"protected/v", wantExplicitResourceDeny(user.conf.awsID, actS3DeleteObjectVersion, objectArn(bucket, "protected/v"))},
+			{"secret/z", wantImplicitDeny(user.arn, actS3DeleteObject, objectArn(bucket, "secret/z"))},
+			{"other/v", wantImplicitDeny(user.arn, actS3DeleteObjectVersion, objectArn(bucket, "other/v"))},
+		})
+	})
+}
+
 // S3IAMAccessControl_governance_bypass_delete_objects verifies the batch
 // DeleteObjects path enforces the bypass permission per object, the same way
 // the single-object delete does.

tests/integration/s3_iam_utils.go

@@ -326,20 +326,35 @@ func runS3ConditionCases(root *iam.Client, s *S3Conf, bucket, key string, cases
 }
 
 func deleteObjectsWithBypass(client *s3.Client, bucket string, keys ...string) (*s3.DeleteObjectsOutput, error) {
-	objects := make([]types.ObjectIdentifier, len(keys))
-	for i, key := range keys {
-		objects[i] = types.ObjectIdentifier{Key: aws.String(key)}
-	}
-
 	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
 	defer cancel()
 	return client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
 		Bucket:                    &bucket,
-		Delete:                    &types.Delete{Objects: objects},
+		Delete:                    &types.Delete{Objects: objectIdentifiers(keys...)},
 		BypassGovernanceRetention: aws.Bool(true),
 	})
 }
 
+// deleteObjectsBatch deletes keys in one DeleteObjects request, with no
+// governance-bypass header, for the tests measuring authorization alone.
+func deleteObjectsBatch(client *s3.Client, bucket string, keys ...string) (*s3.DeleteObjectsOutput, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	defer cancel()
+	return client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
+		Bucket: &bucket,
+		Delete: &types.Delete{Objects: objectIdentifiers(keys...)},
+	})
+}
+
+// objectIdentifiers names keys as unversioned DeleteObjects entries.
+func objectIdentifiers(keys ...string) []types.ObjectIdentifier {
+	objects := make([]types.ObjectIdentifier, len(keys))
+	for i, key := range keys {
+		objects[i] = types.ObjectIdentifier{Key: aws.String(key)}
+	}
+	return objects
+}
+
 // checkDeleteObjectsErr checks one DeleteObjects response entry against the
 // key and denial it's expected to carry.
 func checkDeleteObjectsErr(got types.Error, wantKey string, wantErr s3err.S3Error) error {
@@ -360,12 +375,18 @@ func checkDeleteObjectsErr(got types.Error, wantKey string, wantErr s3err.S3Erro
 // list names exactly wantKeys, in that order — DeleteObjects preserves the
 // order objects were requested in across both the Deleted and Error lists.
 func checkDeletedKeysInOrder(got []types.DeletedObject, wantKeys []string) error {
-	if len(got) != len(wantKeys) {
-		return fmt.Errorf("expected %d deleted objects %v, got %+v", len(wantKeys), wantKeys, got)
+	gotKeys := make([]string, len(got))
+	for i, obj := range got {
+		if obj.Key != nil {
+			gotKeys[i] = *obj.Key
+		}
+	}
+	if len(gotKeys) != len(wantKeys) {
+		return fmt.Errorf("expected %d deleted objects %q, got %d: %q", len(wantKeys), wantKeys, len(gotKeys), gotKeys)
 	}
 	for i, want := range wantKeys {
-		if got[i].Key == nil || *got[i].Key != want {
-			return fmt.Errorf("expected deleted object %d to be %q, got %+v", i, want, got)
+		if gotKeys[i] != want {
+			return fmt.Errorf("expected deleted object %d to be %q, got %q", i, want, gotKeys)
 		}
 	}
 	return nil

tests/integration/utils.go

@@ -3826,13 +3826,17 @@ func hexBytes(s string) string {
 	return strings.Join(parts, " ")
 }
 
+// keyDenial pairs a DeleteObjects key with the error its response entry is
+// expected to carry.
+type keyDenial struct {
+	key string
+	err s3err.S3Error
+}
+
 // checkDeleteObjectsErrsInOrder checks that got names exactly the (key,
 // error) pairs in want, in that order — DeleteObjects preserves the order
 // objects were requested in across both the Deleted and Error lists.
-func checkDeleteObjectsErrsInOrder(got []types.Error, want []struct {
-	key string
-	err s3err.S3Error
-}) error {
+func checkDeleteObjectsErrsInOrder(got []types.Error, want []keyDenial) error {
 	if len(got) != len(want) {
 		return fmt.Errorf("expected %d per-object errors, got %d: %+v", len(want), len(got), got)
 	}