S3: head_object() should only return 405 on DeleteMarkers when specifying a Versionid (#6922)

moto/s3/exceptions.py

@@ -1,6 +1,10 @@
-from typing import Any, Optional, Union
+from typing import Any, Optional, Union, TYPE_CHECKING
 from moto.core.exceptions import RESTError
 
+if TYPE_CHECKING:
+    from moto.s3.models import FakeDeleteMarker
+
+
 ERROR_WITH_BUCKET_NAME = """{% extends 'single_error' %}
 {% block extra %}<BucketName>{{ bucket }}</BucketName>{% endblock %}
 """
@@ -561,8 +565,8 @@ class ObjectLockConfigurationNotFoundError(S3ClientError):
         )
 
 
-class MethodNotAllowed(S3ClientError):
+class HeadOnDeleteMarker(Exception):
-    code = 405
+    """Marker to indicate that we've called `head_object()` on a FakeDeleteMarker"""
 
-    def __init__(self) -> None:
+    def __init__(self, marker: "FakeDeleteMarker"):
-        super().__init__("MethodNotAllowed", "Method Not Allowed")
+        self.marker = marker

moto/s3/models.py

@@ -42,7 +42,7 @@ from moto.s3.exceptions import (
     MissingKey,
     InvalidNotificationDestination,
     MalformedXML,
-    MethodNotAllowed,
+    HeadOnDeleteMarker,
     InvalidStorageClass,
     InvalidTargetBucketForLogging,
     CrossLocationLoggingProhibitted,
@@ -2107,7 +2107,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
             bucket_name, key_name, version_id, part_number, return_delete_marker=True
         )
         if isinstance(obj, FakeDeleteMarker):
-            raise MethodNotAllowed
+            raise HeadOnDeleteMarker(obj)
         return obj
 
     def get_object_acl(self, key: FakeKey) -> Optional[FakeAcl]:

moto/s3/responses.py

@@ -34,7 +34,7 @@ from .exceptions import (
     InvalidContentMD5,
     InvalidContinuationToken,
     S3ClientError,
-    MethodNotAllowed,
+    HeadOnDeleteMarker,
     MissingBucket,
     MissingKey,
     MissingVersion,
@@ -1766,14 +1766,18 @@ class S3Response(BaseResponse):
             key = self.backend.head_object(
                 bucket_name, key_name, version_id=version_id, part_number=part_number
             )
-        except MethodNotAllowed:
+        except HeadOnDeleteMarker as exc:
             headers = {
                 "x-amz-delete-marker": "true",
                 "x-amz-version-id": version_id,
-                "allow": "DELETE",
                 "content-type": "application/xml",
             }
-            return 405, headers, "Method Not Allowed"
+            if version_id:
+                headers["allow"] = "DELETE"
+                return 405, headers, "Method Not Allowed"
+            else:
+                headers["x-amz-version-id"] = exc.marker.version_id
+                return 404, headers, "Not Found"
         if key:
             response_headers.update(key.metadata)
             response_headers.update(key.response_dict)

tests/test_s3/test_s3.py

@@ -1708,7 +1708,7 @@ def test_delete_versioned_bucket_returns_metadata(name=None):
 
     # list_object_versions returns the object itself, and a DeleteMarker
     # object.head() returns a 'x-amz-delete-marker' header
-    # delete_marker.head() returns a 405
+    # delete_marker_version.head() returns a 405
     for version in versions.filter(Prefix="test1"):
         if version.version_id == deleted_version_id:
             with pytest.raises(ClientError) as exc:
@@ -1723,6 +1723,18 @@ def test_delete_versioned_bucket_returns_metadata(name=None):
         else:
             assert version.head()["ResponseMetadata"]["HTTPStatusCode"] == 200
 
+    # Note that delete_marker.head() returns a regular 404
+    # i.e., without specifying the versionId
+    with pytest.raises(ClientError) as exc:
+        client.head_object(Bucket=name, Key="test1")
+    err = exc.value.response
+    assert err["Error"] == {"Code": "404", "Message": "Not Found"}
+    assert err["ResponseMetadata"]["HTTPStatusCode"] == 404
+    assert err["ResponseMetadata"]["HTTPHeaders"]["x-amz-delete-marker"] == "true"
+    assert (
+        err["ResponseMetadata"]["HTTPHeaders"]["x-amz-version-id"] == deleted_version_id
+    )
+
     # Delete the same object gives a new version id
     del_mrk1 = client.delete_object(Bucket=name, Key="test1")
     assert del_mrk1["DeleteMarker"] is True