Repository: getmoto/moto (python)

Issue:
S3 HeadObject does not support range header
The real S3 API supports the `range` header on a HeadObject request, even though the response body is empty. Some clients, like TransferManager used by tensorflow-io, send HeadObject requests with a range header to determine the content length up front. But when used with moto server they will receive a 416.

##  MRE

Run moto server on port 3000:
```
moto_server -p3000
```

Create bucket, object and head object request:
```python
import boto3
import io

# Create a mock S3 client
s3 = boto3.client('s3',
                  endpoint_url='http://localhost:3000', aws_access_key_id="foo", aws_secret_access_key="bar")

# Create a bucket
bucket_name = 'my-tfrecord-bucket'
s3.create_bucket(Bucket=bucket_name)

## Put object
response = s3.put_object(
    Bucket='my-tfrecord-bucket',
    Key='hello',
    Body='world'
)

## Head object request with range
s3.head_object(
    Bucket='my-tfrecord-bucket',
    Key='hello',
    Range='bytes=0-9'
)
```

Errors with:
```
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "../.venv/lib/python3.11/site-packages/botocore/client.py", line 565, in _api_call
    return self._make_api_call(operation_name, kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "../.venv/lib/python3.11/site-packages/botocore/client.py", line 1017, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (416) when calling the HeadObject operation: Requested Range Not Satisfiable
```

moto 5.0.13

Pull request description:
S3: head_object() now handles Range-parameter correctly
Fixes #8047 

Patch:
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index 684145cbf..9a4bef943 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -881,9 +881,11 @@ class S3Response(BaseResponse):
             # - you should not use any location constraint
             location_constraint = self._get_location_constraint()
             if self.region == DEFAULT_REGION_NAME:
-                # REGION = us-east-1 - we should never receive a LocationConstraint
-                if location_constraint:
+                # REGION = us-east-1 - we should never receive a LocationConstraint for `us-east-1`
+                # However, you can create buckets in other regions from us-east-1
+                if location_constraint == DEFAULT_REGION_NAME:
                     raise InvalidLocationConstraintException
+
             else:
                 # Non-Standard region - the LocationConstraint must be equal to the actual region the request was send to
                 if not location_constraint:
@@ -893,8 +895,11 @@ class S3Response(BaseResponse):
             if self.body and not location_constraint:
                 raise MalformedXML()
 
+            bucket_region = (
+                location_constraint if location_constraint else DEFAULT_REGION_NAME
+            )
             try:
-                new_bucket = self.backend.create_bucket(bucket_name, self.region)
+                new_bucket = self.backend.create_bucket(bucket_name, bucket_region)
             except BucketAlreadyExists:
                 new_bucket = self.backend.get_bucket(bucket_name)
                 if new_bucket.account_id == self.get_current_account():
@@ -1270,7 +1275,10 @@ class S3Response(BaseResponse):
     def _handle_range_header(
         self, request: Any, response_headers: Dict[str, Any], response_content: Any
     ) -> TYPE_RESPONSE:
-        length = len(response_content)
+        if request.method == "HEAD":
+            length = int(response_headers["content-length"])
+        else:
+            length = len(response_content)
         last = length - 1
 
         _, rspec = request.headers.get("range").split("=")
@@ -1284,6 +1292,8 @@ class S3Response(BaseResponse):
             return 200, response_headers, response_content
 
         if (begin is None and end == 0) or (begin is not None and begin > last):
+            if request.method == "HEAD":
+                return 416, {"Content-Type": "application/xml"}, b""
             raise InvalidRange(
                 actual_size=str(length), range_requested=request.headers.get("range")
             )
@@ -1303,7 +1313,10 @@ class S3Response(BaseResponse):
 
         response_headers["content-range"] = f"bytes {begin}-{end}/{length}"
         content = response_content[begin : end + 1]
-        response_headers["content-length"] = str(len(content))
+        if request.method == "HEAD":
+            response_headers["content-length"] = str((end - begin) + 1)
+        else:
+            response_headers["content-length"] = str(len(content))
         return 206, response_headers, content
 
     def _handle_v4_chunk_signatures(self, body: bytes, content_length: int) -> bytes:
@@ -1321,12 +1334,20 @@ class S3Response(BaseResponse):
             line = body_io.readline()
         return bytes(new_body)
 
-    def _handle_encoded_body(self, body: bytes, content_length: int) -> bytes:
+    def _handle_encoded_body(self, body: bytes) -> bytes:
+        decoded_body = b""
         body_io = io.BytesIO(body)
-        # first line should equal '{content_length}\r\n
-        body_io.readline()
-        # Body contains actual data next
-        return body_io.read(content_length)
+        # first line should equal '{content_length}\r\n' while the content_length is a hex number
+        content_length = int(body_io.readline().strip(), 16)
+        while content_length > 0:
+            # read the content_length of the actual data
+            decoded_body += body_io.read(content_length)
+            # next is line with just '\r\n' so we skip it
+            body_io.readline()
+            # read another line with '{content_length}\r\n'
+            content_length = int(body_io.readline().strip(), 16)
+
+        return decoded_body
         # last line should equal
         # amz-checksum-sha256:<..>\r\n
 
@@ -1685,9 +1706,7 @@ class S3Response(BaseResponse):
             request.headers.get("x-amz-content-sha256", None)
             == "STREAMING-UNSIGNED-PAYLOAD-TRAILER"
         ):
-            body = self._handle_encoded_body(
-                body, int(request.headers["x-amz-decoded-content-length"])
-            )
+            body = self._handle_encoded_body(body)
 
         bucket_key_enabled = request.headers.get(
             "x-amz-server-side-encryption-bucket-key-enabled", None


Test patch:
diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py
index c9195d0e9..6bb8539e8 100644
--- a/tests/test_s3/test_s3.py
+++ b/tests/test_s3/test_s3.py
@@ -292,6 +292,22 @@ def test_create_bucket_with_wrong_location_constraint(constraint):
     )
 
 
+@aws_verified
+@pytest.mark.aws_verified
+@pytest.mark.parametrize("constraint", ["us-east-2", "eu-central-1"])
+def test_create_bucket_in_regions_from_us_east_1(constraint):
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    bucket_name = str(uuid.uuid4())
+    client.create_bucket(
+        Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": constraint}
+    )
+    assert (
+        client.get_bucket_location(Bucket=bucket_name)["LocationConstraint"]
+        == constraint
+    )
+    client.delete_bucket(Bucket=bucket_name)
+
+
 @mock_aws
 def test_create_existing_bucket_in_us_east_1():
     """Creating a bucket that already exists in us-east-1 returns the bucket.
@@ -909,6 +925,22 @@ def test_upload_file_with_checksum_algorithm():
     assert random_bytes == actual_content
 
 
+@mock_aws
+def test_put_large_with_checksum_algorithm():
+    s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    bucket = "mybucket"
+    s3_client.create_bucket(Bucket=bucket)
+    s3_client.put_object(
+        Bucket=bucket,
+        Key="the-key",
+        Body=b"test" * 1000000,
+        ChecksumAlgorithm="SHA256",
+    )
+
+    resp = s3_client.get_object(Bucket="mybucket", Key="the-key")
+    assert resp["Body"].read() == b"test" * 1000000
+
+
 @mock_aws
 def test_put_chunked_with_v4_signature_in_body():
     bucket_name = "mybucket"
@@ -3391,6 +3423,41 @@ def test_head_versioned_key_in_not_versioned_bucket():
     assert response["Error"]["Code"] == "400"
 
 
+@s3_aws_verified
+@pytest.mark.aws_verified
+def test_head_object_with_range_header(bucket_name=None):
+    # HeadObject returns only the metadata for an object.
+    # If the Range is satisfiable, only the ContentLength is affected in the response.
+    # If the Range is not satisfiable, S3 returns a 416 - Requested Range Not Satisfiable error.
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    client.put_object(Bucket=bucket_name, Key="f1", Body="f1")
+
+    for valid in ["0-9", "0-"]:
+        resp = client.head_object(Bucket=bucket_name, Key="f1", Range=f"bytes={valid}")
+        headers = resp["ResponseMetadata"]["HTTPHeaders"]
+        assert headers["content-range"] == "bytes 0-1/2"
+        assert headers["content-length"] == "2"
+
+    for valid in ["-1", "1-1"]:
+        resp = client.head_object(Bucket=bucket_name, Key="f1", Range=f"bytes={valid}")
+        headers = resp["ResponseMetadata"]["HTTPHeaders"]
+        assert headers["content-range"] == "bytes 1-1/2"
+        assert headers["content-length"] == "1"
+
+    for valid in ["1-0"]:
+        resp = client.head_object(Bucket=bucket_name, Key="f1", Range=f"bytes={valid}")
+        headers = resp["ResponseMetadata"]["HTTPHeaders"]
+        assert "content-range" not in headers
+        assert headers["content-length"] == "2"
+
+    for invalid in ["5-9", "100-"]:
+        with pytest.raises(ClientError) as exc:
+            client.head_object(Bucket=bucket_name, Key="f1", Range=f"bytes={invalid}")
+        err = exc.value.response["Error"]
+        assert err["Code"] == "416"
+        assert "Range Not Satisfiable" in err["Message"]
+
+
 @mock_aws
 def test_prefix_encoding():
     bucket_name = "encoding-bucket"