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

objectsAccessErrors recorded the first resource-policy Deny and returned immediately, leaving every later key in the same action subset with a nil result — which VerifyObjectsAccess caller reads as "authorized" and sends straight to the backend. An explicit bucket-policy Deny therefore let the keys after it skip authorization entirely and be deleted, including keys that same bucket policy explicitly denied.

The loop now continues rather than returning, so every denied key is settled with its own error. The identity-policy round trip the early return was saving is still skipped, but only when the bucket policy denied every key in the batch, since no identity-policy answer could change any result then. Both loops that follow skip keys already holding an error, so a resource-level explicit deny is never overwritten by an identity-policy result nor flattened to the generic AccessDenied message.

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)
 	}