Repository: getmoto/moto
Pull request #8054 (2024-08-29 18:06:35)

ACM InUseBy with elbv2
This PR is similar to https://github.com/getmoto/moto/pull/4414, implementing the same use case for ELBV2.
In a nutshell, it updates the `InUseBy` property of an ACM certificate when attaching it to an ELBV2 listener.

I also fixed the `ListCertificate` operation that should return a boolean `InUse` property, which gets computed from the `InUseBy` attached to the certificate description.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/8054?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 94.43%. Comparing base [(`191ebcc`)](https://app.codecov.io/gh/getmoto/moto/commit/191ebcc02f7d63db3efdc7ea40f37c10a86bd7c7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`8122e93`)](https://app.codecov.io/gh/getmoto/moto/commit/8122e935c6b63e0c1ff7fd372e5113ba01d2aba9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##           master    #8054   +/-   ##
=======================================
  Coverage   94.43%   94.43%           
=======================================
  Files        1139     1139           
  Lines       97355    97360    +5     
=======================================
+ Hits        91935    91943    +8     
+ Misses       5420     5417    -3     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/8054/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/8054/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `28.92% <18.75%> (+<0.01%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/8054/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `94.40% <100.00%> (+<0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/8054?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).


Patch:
diff --git a/moto/acm/responses.py b/moto/acm/responses.py
index 9d5daa68b..97c27cae1 100644
--- a/moto/acm/responses.py
+++ b/moto/acm/responses.py
@@ -130,7 +130,10 @@ class AWSCertificateManagerResponse(BaseResponse):
         certs = []
         statuses = self._get_param("CertificateStatuses")
         for cert_bundle in self.acm_backend.list_certificates(statuses):
-            certs.append(cert_bundle.describe()["Certificate"])
+            _cert = cert_bundle.describe()["Certificate"]
+            _in_use_by = _cert.pop("InUseBy", [])
+            _cert["InUse"] = bool(_in_use_by)
+            certs.append(_cert)
 
         result = {"CertificateSummaryList": certs}
         return json.dumps(result)
diff --git a/moto/elb/models.py b/moto/elb/models.py
index 83dc01c3a..c186dde05 100644
--- a/moto/elb/models.py
+++ b/moto/elb/models.py
@@ -85,7 +85,6 @@ class FakeLoadBalancer(CloudFormationModel):
         vpc_id: Optional[str],
         subnets: Optional[List[str]],
         security_groups: Optional[List[str]],
-        elb_backend: "ELBBackend",
     ):
         self.name = name
         self.health_check: Optional[FakeHealthCheck] = None
@@ -115,13 +114,6 @@ class FakeLoadBalancer(CloudFormationModel):
                     "ssl_certificate_id", port.get("SSLCertificateId")
                 ),
             )
-            if listener.ssl_certificate_id and not re.search(
-                f"{ARN_PARTITION_REGEX}:iam:", listener.ssl_certificate_id
-            ):
-                elb_backend._register_certificate(
-                    listener.ssl_certificate_id, self.dns_name
-                )
-
             self.listeners.append(listener)
 
             # it is unclear per the AWS documentation as to when or how backend
@@ -345,8 +337,17 @@ class ELBBackend(BaseBackend):
             subnets=subnets,
             security_groups=security_groups,
             vpc_id=vpc_id,
-            elb_backend=self,
         )
+        for listener in new_load_balancer.listeners:
+            if listener.ssl_certificate_id and not re.search(
+                f"{ARN_PARTITION_REGEX}:iam:", listener.ssl_certificate_id
+            ):
+                register_certificate(
+                    self.account_id,
+                    self.region_name,
+                    listener.ssl_certificate_id,
+                    new_load_balancer.dns_name,
+                )
         self.load_balancers[name] = new_load_balancer
         return new_load_balancer
 
@@ -373,8 +374,11 @@ class ELBBackend(BaseBackend):
                     if ssl_certificate_id and not re.search(
                         f"{ARN_PARTITION_REGEX}:iam::", ssl_certificate_id
                     ):
-                        self._register_certificate(
-                            ssl_certificate_id, balancer.dns_name
+                        register_certificate(
+                            self.account_id,
+                            self.region_name,
+                            ssl_certificate_id,
+                            balancer.dns_name,
                         )
                     balancer.listeners.append(
                         FakeListener(
@@ -493,8 +497,11 @@ class ELBBackend(BaseBackend):
                 if lb_port == listener.load_balancer_port:
                     balancer.listeners[idx].ssl_certificate_id = ssl_certificate_id
                     if ssl_certificate_id:
-                        self._register_certificate(
-                            ssl_certificate_id, balancer.dns_name
+                        register_certificate(
+                            self.account_id,
+                            self.region_name,
+                            ssl_certificate_id,
+                            balancer.dns_name,
                         )
         return balancer
 
@@ -621,15 +628,6 @@ class ELBBackend(BaseBackend):
         load_balancer.listeners[listener_idx] = listener
         return load_balancer
 
-    def _register_certificate(self, ssl_certificate_id: str, dns_name: str) -> None:
-        from moto.acm.models import CertificateNotFound, acm_backends
-
-        acm_backend = acm_backends[self.account_id][self.region_name]
-        try:
-            acm_backend.set_certificate_in_use_by(ssl_certificate_id, dns_name)
-        except CertificateNotFound:
-            raise CertificateNotFoundException()
-
     def enable_availability_zones_for_load_balancer(
         self, load_balancer_name: str, availability_zones: List[str]
     ) -> List[str]:
@@ -665,4 +663,16 @@ class ELBBackend(BaseBackend):
         return load_balancer.subnets
 
 
+def register_certificate(
+    account_id: str, region: str, arn_certificate: str, arn_user: str
+) -> None:
+    from moto.acm.models import CertificateNotFound, acm_backends
+
+    acm_backend = acm_backends[account_id][region]
+    try:
+        acm_backend.set_certificate_in_use_by(arn_certificate, arn_user)
+    except CertificateNotFound:
+        raise CertificateNotFoundException()
+
+
 elb_backends = BackendDict(ELBBackend, "elb")
diff --git a/moto/elbv2/models.py b/moto/elbv2/models.py
index 73bfdb468..ba4e2bef6 100644
--- a/moto/elbv2/models.py
+++ b/moto/elbv2/models.py
@@ -14,6 +14,8 @@ from moto.ec2.models.subnets import Subnet
 from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
 
+from ..elb.models import register_certificate
+from ..utilities.utils import ARN_PARTITION_REGEX
 from .exceptions import (
     ActionTargetGroupNotFoundError,
     DuplicateListenerError,
@@ -1420,6 +1422,14 @@ Member must satisfy regular expression pattern: {expression}"
             default_actions,
             alpn_policy,
         )
+        if certificate and not re.search(f"{ARN_PARTITION_REGEX}:iam:", certificate):
+            register_certificate(
+                account_id=self.account_id,
+                region=self.region_name,
+                arn_certificate=certificate,
+                arn_user=arn,
+            )
+
         balancer.listeners[listener.arn] = listener
         for action in default_actions:
             if action.type == "forward":
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_acm/test_acm.py b/tests/test_acm/test_acm.py
index 91e75e436..1a22fbc27 100644
--- a/tests/test_acm/test_acm.py
+++ b/tests/test_acm/test_acm.py
@@ -9,6 +9,7 @@ from freezegun import freeze_time
 
 from moto import mock_aws, settings
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
+from tests.test_elbv2.test_elbv2 import create_load_balancer
 
 RESOURCE_FOLDER = os.path.join(os.path.dirname(__file__), "resources")
 
@@ -710,7 +711,7 @@ def test_request_certificate_issued_status_with_wait_in_envvar():
 
 
 @mock_aws
-def test_request_certificate_with_mutiple_times():
+def test_request_certificate_with_multiple_times():
     if not settings.TEST_DECORATOR_MODE:
         raise SkipTest("Cant manipulate time in server mode")
 
@@ -789,3 +790,44 @@ def test_elb_acm_in_use_by():
     assert response["Certificate"]["InUseBy"] == [
         create_load_balancer_request["DNSName"]
     ]
+
+    certificates = acm_client.list_certificates()["CertificateSummaryList"]
+    cert = [
+        _cert for _cert in certificates if _cert["CertificateArn"] == certificate_arn
+    ][0]
+    assert cert["InUse"]
+
+
+@mock_aws
+def test_elbv2_acm_in_use_by():
+    acm_client = boto3.client("acm", region_name="us-east-1")
+
+    certificate_arn = acm_client.request_certificate(
+        DomainName="fake.domain.com",
+        DomainValidationOptions=[
+            {"DomainName": "fake.domain.com", "ValidationDomain": "domain.com"}
+        ],
+    )["CertificateArn"]
+
+    response, _, _, _, _, elbv2_client = create_load_balancer()
+    load_balancer_arn = response["LoadBalancers"][0]["LoadBalancerArn"]
+
+    create_listener_resp = elbv2_client.create_listener(
+        LoadBalancerArn=load_balancer_arn,
+        Protocol="HTTP",
+        Port=443,
+        DefaultActions=[],
+        Certificates=[{"CertificateArn": certificate_arn}],
+        AlpnPolicy=["pol1", "pol2"],
+    )
+
+    response = acm_client.describe_certificate(CertificateArn=certificate_arn)
+    assert response["Certificate"]["InUseBy"] == [
+        create_listener_resp["Listeners"][0]["ListenerArn"]
+    ]
+
+    certificates = acm_client.list_certificates()["CertificateSummaryList"]
+    cert = [
+        _cert for _cert in certificates if _cert["CertificateArn"] == certificate_arn
+    ][0]
+    assert cert["InUse"]
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"