Repository: getmoto/moto
Pull request #8039 (2024-08-27 12:42:21)

ACM-PCA: Use serialisable types instead of native objs
This is a follow up to https://github.com/getmoto/moto/pull/7958

The `ImportCertificateAuthorityCertificate` would create a `cryptography.hazmat.bindings._rust.x509.Certificate` object as assign it as a `CertificateAuthority` class member. This prevents it from being serialised with pickle.

This PR reworks the certificate and private key attributes so that all members are picklable.

bblommers: Hey @viren-nadkarni! Do you want to add some tests for this as well in Moto? Just to prevent it from breaking again in the future
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/8039?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.41%. Comparing base [(`9d7a102`)](https://app.codecov.io/gh/getmoto/moto/commit/9d7a102464ca4ce5e0aa8e27cd53866e8c4aa10d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`0dd3bd7`)](https://app.codecov.io/gh/getmoto/moto/commit/0dd3bd7b5087c8fe39a7dbb5c35d76ab6e73d1ea?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).
> Report is 11 commits behind head on master.

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


```diff
@@           Coverage Diff           @@
##           master    #8039   +/-   ##
=======================================
  Coverage   94.41%   94.41%           
=======================================
  Files        1134     1134           
  Lines       96764    96832   +68     
=======================================
+ Hits        91356    91425   +69     
+ Misses       5408     5407    -1     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/8039/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/8039/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `28.99% <0.00%> (-0.02%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/8039/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `94.38% <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/8039?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/acmpca/exceptions.py b/moto/acmpca/exceptions.py
index b17c9b245..674d2d89e 100644
--- a/moto/acmpca/exceptions.py
+++ b/moto/acmpca/exceptions.py
@@ -26,3 +26,13 @@ class InvalidStateException(JsonRESTError):
             "InvalidStateException",
             f"The certificate authority {arn} is not in the correct state to have a certificate signing request.",
         )
+
+
+class MalformedCertificateAuthorityException(JsonRESTError):
+    code = 400
+
+    def __init__(self) -> None:
+        super().__init__(
+            "MalformedCertificateAuthorityException",
+            "Malformed certificate.",
+        )
diff --git a/moto/acmpca/models.py b/moto/acmpca/models.py
index 38d91f9f8..bcff2ffa6 100644
--- a/moto/acmpca/models.py
+++ b/moto/acmpca/models.py
@@ -2,7 +2,7 @@
 
 import base64
 import datetime
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, cast
 
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
@@ -19,6 +19,7 @@ from moto.utilities.utils import get_partition
 from .exceptions import (
     InvalidS3ObjectAclInCrlConfiguration,
     InvalidStateException,
+    MalformedCertificateAuthorityException,
     ResourceNotFoundException,
 )
 
@@ -49,14 +50,16 @@ class CertificateAuthority(BaseModel):
         self.usage_mode = "SHORT_LIVED_CERTIFICATE"
         self.security_standard = security_standard or "FIPS_140_2_LEVEL_3_OR_HIGHER"
 
-        self.key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
         self.password = str(mock_random.uuid4()).encode("utf-8")
-        self.private_bytes = self.key.private_bytes(
+
+        private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        self.private_bytes = private_key.private_bytes(
             encoding=serialization.Encoding.PEM,
             format=serialization.PrivateFormat.TraditionalOpenSSL,
             encryption_algorithm=serialization.BestAvailableEncryption(self.password),
         )
-        self.certificate: Optional[x509.Certificate] = None
+
+        self.certificate_bytes: bytes = b""
         self.certificate_chain: Optional[bytes] = None
         self.issued_certificates: Dict[str, bytes] = dict()
 
@@ -85,6 +88,20 @@ class CertificateAuthority(BaseModel):
 
         return cert.public_bytes(serialization.Encoding.PEM)
 
+    @property
+    def key(self) -> rsa.RSAPrivateKey:
+        private_key = serialization.load_pem_private_key(
+            self.private_bytes,
+            password=self.password,
+        )
+        return cast(rsa.RSAPrivateKey, private_key)
+
+    @property
+    def certificate(self) -> Optional[x509.Certificate]:
+        if self.certificate_bytes:
+            return x509.load_pem_x509_certificate(self.certificate_bytes)
+        return None
+
     @property
     def issuer(self) -> x509.Name:
         name_attributes = []
@@ -260,12 +277,6 @@ class CertificateAuthority(BaseModel):
                     if acl not in ["PUBLIC_READ", "BUCKET_OWNER_FULL_CONTROL"]:
                         raise InvalidS3ObjectAclInCrlConfiguration(acl)
 
-    @property
-    def certificate_bytes(self) -> bytes:
-        if self.certificate:
-            return self.certificate.public_bytes(serialization.Encoding.PEM)
-        return b""
-
     @property
     def not_valid_after(self) -> Optional[float]:
         if self.certificate is None:
@@ -287,7 +298,12 @@ class CertificateAuthority(BaseModel):
     def import_certificate_authority_certificate(
         self, certificate: bytes, certificate_chain: Optional[bytes]
     ) -> None:
-        self.certificate = x509.load_pem_x509_certificate(certificate)
+        try:
+            x509.load_pem_x509_certificate(certificate)
+        except ValueError:
+            raise MalformedCertificateAuthorityException()
+
+        self.certificate_bytes = certificate
         self.certificate_chain = certificate_chain
         self.status = "ACTIVE"
         self.updated_at = unix_time()
diff --git a/moto/apigateway/responses.py b/moto/apigateway/responses.py
index 8434722ae..127128419 100644
--- a/moto/apigateway/responses.py
+++ b/moto/apigateway/responses.py
@@ -1,5 +1,5 @@
 import json
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 from urllib.parse import unquote
 
 from moto.core.responses import TYPE_RESPONSE, BaseResponse
@@ -28,7 +28,7 @@ class APIGatewayResponse(BaseResponse):
     def backend(self) -> APIGatewayBackend:
         return apigateway_backends[self.current_account][self.region]
 
-    def __validate_api_key_source(self, api_key_source: str) -> TYPE_RESPONSE:  # type: ignore[return]
+    def __validate_api_key_source(self, api_key_source: str) -> Optional[TYPE_RESPONSE]:
         if api_key_source and api_key_source not in API_KEY_SOURCES:
             return self.error(
                 "ValidationException",
@@ -39,10 +39,11 @@ class APIGatewayResponse(BaseResponse):
                     "[AUTHORIZER, HEADER]"
                 ).format(api_key_source=api_key_source),
             )
+        return None
 
-    def __validate_endpoint_configuration(  # type: ignore[return]
+    def __validate_endpoint_configuration(
         self, endpoint_configuration: Dict[str, str]
-    ) -> TYPE_RESPONSE:
+    ) -> Optional[TYPE_RESPONSE]:
         if endpoint_configuration and "types" in endpoint_configuration:
             invalid_types = list(
                 set(endpoint_configuration["types"]) - set(ENDPOINT_CONFIGURATION_TYPES)
@@ -57,6 +58,7 @@ class APIGatewayResponse(BaseResponse):
                         "[PRIVATE, EDGE, REGIONAL]"
                     ).format(endpoint_type=invalid_types[0]),
                 )
+        return None
 
     def create_rest_api(self) -> TYPE_RESPONSE:
         api_doc = deserialize_body(self.body)
@@ -102,14 +104,15 @@ class APIGatewayResponse(BaseResponse):
         apis = self.backend.list_apis()
         return json.dumps({"item": [api.to_dict() for api in apis]})
 
-    def __validte_rest_patch_operations(  # type: ignore[return]
+    def __validte_rest_patch_operations(
         self, patch_operations: List[Dict[str, str]]
-    ) -> TYPE_RESPONSE:
+    ) -> Optional[TYPE_RESPONSE]:
         for op in patch_operations:
             path = op["path"]
             if "apiKeySource" in path:
                 value = op["value"]
                 return self.__validate_api_key_source(value)
+        return None
 
     def delete_rest_api(self) -> TYPE_RESPONSE:
         function_id = self.path.replace("/restapis/", "", 1).split("/")[0]
@@ -121,7 +124,8 @@ class APIGatewayResponse(BaseResponse):
         rest_api = self.backend.get_rest_api(function_id)
         return 200, {}, json.dumps(rest_api.to_dict())
 
-    def get_rest_api_without_id(self, *args: Any) -> TYPE_RESPONSE:  # type: ignore[misc]
+    @staticmethod
+    def get_rest_api_without_id(*args: Any) -> TYPE_RESPONSE:  # type: ignore[misc]
         """
         AWS is returning an empty response when restApiId is an empty string. This is slightly odd and it seems an
         outlier, therefore it was decided we could have a custom handler for this particular use case instead of
diff --git a/moto/apigateway/urls.py b/moto/apigateway/urls.py
index 49d291611..9bcd317af 100644
--- a/moto/apigateway/urls.py
+++ b/moto/apigateway/urls.py
@@ -44,4 +44,4 @@ url_paths = {
 }
 
 # Also manages the APIGatewayV2
-url_paths.update(url_paths_v2)  # type: ignore[arg-type]
+url_paths.update(url_paths_v2)
diff --git a/moto/awslambda/responses.py b/moto/awslambda/responses.py
index 4f825b991..6c2328025 100644
--- a/moto/awslambda/responses.py
+++ b/moto/awslambda/responses.py
@@ -321,20 +321,14 @@ class LambdaResponse(BaseResponse):
 
     def get_function_concurrency(self) -> TYPE_RESPONSE:
         path_function_name = unquote(self.path.rsplit("/", 2)[-2])
-        function_name = self.backend.get_function(path_function_name)
-
-        if function_name is None:
-            return 404, {"status": 404}, "{}"
+        self.backend.get_function(path_function_name)
 
         resp = self.backend.get_function_concurrency(path_function_name)
         return 200, {}, json.dumps({"ReservedConcurrentExecutions": resp})
 
     def delete_function_concurrency(self) -> TYPE_RESPONSE:
         path_function_name = unquote(self.path.rsplit("/", 2)[-2])
-        function_name = self.backend.get_function(path_function_name)
-
-        if function_name is None:
-            return 404, {}, "{}"
+        self.backend.get_function(path_function_name)
 
         self.backend.delete_function_concurrency(path_function_name)
 
@@ -342,10 +336,7 @@ class LambdaResponse(BaseResponse):
 
     def put_function_concurrency(self) -> TYPE_RESPONSE:
         path_function_name = unquote(self.path.rsplit("/", 2)[-2])
-        function = self.backend.get_function(path_function_name)
-
-        if function is None:
-            return 404, {}, "{}"
+        self.backend.get_function(path_function_name)
 
         concurrency = self._get_param("ReservedConcurrentExecutions", None)
         resp = self.backend.put_function_concurrency(path_function_name, concurrency)
diff --git a/moto/cloudwatch/responses.py b/moto/cloudwatch/responses.py
index 6675d0998..40d7d3c9c 100644
--- a/moto/cloudwatch/responses.py
+++ b/moto/cloudwatch/responses.py
@@ -234,7 +234,7 @@ class CloudWatchResponse(BaseResponse):
 
     def delete_dashboards(self) -> Union[str, ERROR_RESPONSE]:
         dashboards = self._get_multi_param("DashboardNames.member")
-        if dashboards is None:
+        if not dashboards:
             return self._error("InvalidParameterValue", "Need at least 1 dashboard")
 
         error = self.cloudwatch_backend.delete_dashboards(dashboards)
diff --git a/moto/core/responses.py b/moto/core/responses.py
index 4b0ae6689..b849fee21 100644
--- a/moto/core/responses.py
+++ b/moto/core/responses.py
@@ -363,6 +363,7 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
 
         if hasattr(self.body, "read"):
             self.body = self.body.read()
+        self.raw_body = self.body
 
         # https://github.com/getmoto/moto/issues/6692
         # Content coming from SDK's can be GZipped for performance reasons
diff --git a/moto/dynamodb/responses.py b/moto/dynamodb/responses.py
index b0141dfba..fb1f9d57b 100644
--- a/moto/dynamodb/responses.py
+++ b/moto/dynamodb/responses.py
@@ -53,28 +53,25 @@ def include_consumed_capacity(
 
             response = f(*args, **kwargs)
 
-            if isinstance(response, str):
-                body = json.loads(response)
+            body = json.loads(response)
 
-                if expected_capacity == "TOTAL":
-                    body["ConsumedCapacity"] = {
-                        "TableName": table_name,
-                        "CapacityUnits": val,
-                    }
-                elif expected_capacity == "INDEXES":
-                    body["ConsumedCapacity"] = {
-                        "TableName": table_name,
-                        "CapacityUnits": val,
-                        "Table": {"CapacityUnits": val},
+            if expected_capacity == "TOTAL":
+                body["ConsumedCapacity"] = {
+                    "TableName": table_name,
+                    "CapacityUnits": val,
+                }
+            elif expected_capacity == "INDEXES":
+                body["ConsumedCapacity"] = {
+                    "TableName": table_name,
+                    "CapacityUnits": val,
+                    "Table": {"CapacityUnits": val},
+                }
+                if index_name:
+                    body["ConsumedCapacity"]["LocalSecondaryIndexes"] = {
+                        index_name: {"CapacityUnits": val}
                     }
-                    if index_name:
-                        body["ConsumedCapacity"]["LocalSecondaryIndexes"] = {
-                            index_name: {"CapacityUnits": val}
-                        }
-
-                return dynamo_json_dump(body)
 
-            return response
+            return dynamo_json_dump(body)
 
         return _wrapper
 
diff --git a/moto/ec2/responses/launch_templates.py b/moto/ec2/responses/launch_templates.py
index 648c36dbf..595c02217 100644
--- a/moto/ec2/responses/launch_templates.py
+++ b/moto/ec2/responses/launch_templates.py
@@ -187,11 +187,7 @@ class LaunchTemplates(EC2BaseResponse):
         ret_versions = []
         if versions and template is not None:
             for v in versions:
-                if str(v).lower() == "$latest" or "$default":
-                    tv = template.get_version(v)
-                else:
-                    tv = template.get_version(int(v))
-                ret_versions.append(tv)
+                ret_versions.append(template.get_version(v))
         elif min_version:
             if max_version:
                 vMax = max_version
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/emr/models.py b/moto/emr/models.py
index 5922d17a2..28df66ab2 100644
--- a/moto/emr/models.py
+++ b/moto/emr/models.py
@@ -555,7 +555,7 @@ class ElasticMapReduceBackend(BaseBackend):
         self.clusters: Dict[str, FakeCluster] = {}
         self.instance_groups: Dict[str, FakeInstanceGroup] = {}
         self.security_configurations: Dict[str, FakeSecurityConfiguration] = {}
-        self.block_public_access_configuration: Dict[str, Any] = {}  # type: ignore [misc]
+        self.block_public_access_configuration: Dict[str, Any] = {}
 
     @property
     def ec2_backend(self) -> Any:  # type: ignore[misc]
diff --git a/moto/fsx/models.py b/moto/fsx/models.py
index 7d6d51a09..554974ec5 100644
--- a/moto/fsx/models.py
+++ b/moto/fsx/models.py
@@ -174,14 +174,13 @@ class FSxBackend(BaseBackend):
         resource = self._get_resource_from_arn(resource_arn)
         resource.tags.extend(tags)
 
-    def _get_resource_from_arn(self, arn: str) -> Any:
+    def _get_resource_from_arn(self, arn: str) -> FileSystem:
         target_resource, target_name = arn.split(":")[-1].split("/")
         try:
-            resource = self.file_systems.get(target_name)  # type: ignore
+            return self.file_systems[target_name]
         except KeyError:
             message = f"Could not find {target_resource} with name {target_name}"
             raise ValueError(message)
-        return resource
 
     def untag_resource(self, resource_arn: str, tag_keys: List[str]) -> None:
         resource = self._get_resource_from_arn(resource_arn)
diff --git a/moto/glue/responses.py b/moto/glue/responses.py
index c1bf3997c..0ff012910 100644
--- a/moto/glue/responses.py
+++ b/moto/glue/responses.py
@@ -34,7 +34,7 @@ class GlueResponse(BaseResponse):
         self.glue_backend.create_database(
             database_name,
             database_input,  # type: ignore[arg-type]
-            self.parameters.get("Tags"),  # type: ignore[arg-type]
+            self.parameters.get("Tags"),
         )
         return ""
 
diff --git a/moto/networkmanager/models.py b/moto/networkmanager/models.py
index e8eee54a3..238b29c8a 100644
--- a/moto/networkmanager/models.py
+++ b/moto/networkmanager/models.py
@@ -266,10 +266,9 @@ class NetworkManagerBackend(BaseBackend):
                     for inner_dict in resources.values()
                     for k, v in inner_dict.items()
                 }
-            resource = resources.get(target_name)  # type: ignore
+            return resources[target_name]
         except (KeyError, ValueError, AttributeError):
             raise ResourceNotFound(arn)
-        return resource
 
     def create_global_network(
         self,
diff --git a/moto/qldb/models.py b/moto/qldb/models.py
index ef2084f93..3b46855a4 100644
--- a/moto/qldb/models.py
+++ b/moto/qldb/models.py
@@ -110,7 +110,7 @@ class QLDBBackend(BaseBackend):
         Optional[datetime],
         Optional[bool],
         Optional[Dict[str, Union[str, datetime]]],
-    ]:  # type: ignore[misc]
+    ]:
         if name not in self.ledgers:
             raise LedgerNotFoundException(name)
         if deletion_protection is not None:
@@ -131,7 +131,7 @@ class QLDBBackend(BaseBackend):
             ledger.get("encryption_description"),
         )
 
-    def _get_ledger_by_resource_arn(self, resource_arn: str) -> Dict[str, Any]:  # type: ignore[misc]
+    def _get_ledger_by_resource_arn(self, resource_arn: str) -> Dict[str, Any]:
         for ledger in self.ledgers.values():
             if ledger.get("resource_arn") == resource_arn:
                 return ledger
diff --git a/moto/qldb/responses.py b/moto/qldb/responses.py
index 47a487f51..16959646f 100644
--- a/moto/qldb/responses.py
+++ b/moto/qldb/responses.py
@@ -78,7 +78,7 @@ class QLDBResponse(BaseResponse):
             permissions_mode,
             deletion_protection,
             kms_key_arn,
-        ) = self.qldb_backend.create_ledger(  # type: ignore[misc]
+        ) = self.qldb_backend.create_ledger(
             name=name,
             tags=tags,
             permissions_mode=permissions_mode,
diff --git a/moto/route53domains/models.py b/moto/route53domains/models.py
index 3bb36cc86..623bcdbe4 100644
--- a/moto/route53domains/models.py
+++ b/moto/route53domains/models.py
@@ -151,7 +151,7 @@ class Route53DomainsBackend(BaseBackend):
 
         return self.__domains[domain_name]
 
-    @paginate(pagination_model=PAGINATION_MODEL)  # type: ignore[misc]
+    @paginate(pagination_model=PAGINATION_MODEL)
     def list_domains(
         self,
         filter_conditions: Optional[List[Dict[str, Any]]] = None,
@@ -197,7 +197,7 @@ class Route53DomainsBackend(BaseBackend):
 
         return domains_to_return
 
-    @paginate(pagination_model=PAGINATION_MODEL)  # type: ignore[misc]
+    @paginate(pagination_model=PAGINATION_MODEL)
     def list_operations(
         self,
         submitted_since_timestamp: Optional[int] = None,
diff --git a/moto/s3/models.py b/moto/s3/models.py
index aa31450d9..5ae966e9d 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -143,7 +143,7 @@ class FakeKey(BaseModel, ManagedState):
         self.last_modified = utcnow()
         self.acl: Optional[FakeAcl] = get_canned_acl("private")
         self.website_redirect_location: Optional[str] = None
-        self.checksum_algorithm = None
+        self.checksum_algorithm: Optional[str] = None
         self._storage_class: Optional[str] = storage if storage else "STANDARD"
         self._metadata = LowercaseDict()
         self._expiry: Optional[datetime.datetime] = None
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index 684145cbf..f8ecf20b3 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -183,6 +183,19 @@ class S3Response(BaseResponse):
             )
         self.bucket_name = self.parse_bucket_name_from_url(request, full_url)
         self.request = request
+        if (
+            self.request.headers.get("x-amz-content-sha256")
+            == "STREAMING-UNSIGNED-PAYLOAD-TRAILER"
+        ):
+            self.body = self._handle_encoded_body(self.body)
+
+        if (
+            request.headers.get("x-amz-content-sha256")
+            == "STREAMING-AWS4-HMAC-SHA256-PAYLOAD"
+        ):
+            self.body = self._handle_v4_chunk_signatures(
+                self.raw_body, int(request.headers["x-amz-decoded-content-length"])
+            )
 
     def get_safe_path(self) -> str:
         return unquote(self.raw_path)
@@ -881,9 +894,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 +908,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 +1288,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 +1305,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 +1326,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 +1347,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
 
@@ -1389,7 +1423,7 @@ class S3Response(BaseResponse):
 
         # Enforces policy when creating new objects, `if key` does not https://github.com/getmoto/moto/issues/7837
         if bucket:
-            resource = f"arn:{bucket.partition}:s3:::{bucket_name}/{key_name}"  # type: ignore[union-attr]
+            resource = f"arn:{bucket.partition}:s3:::{bucket_name}/{key_name}"
 
             # Authorization Workflow
             # https://docs.aws.amazon.com/AmazonS3/latest/userguide/access-control-auth-workflow-object-operation.html
@@ -1398,7 +1432,7 @@ class S3Response(BaseResponse):
             from moto.iam.access_control import PermissionResult
 
             action = f"s3:{method.upper()[0]}{method.lower()[1:]}Object"
-            bucket_permissions = bucket.get_permission(action, resource)  # type: ignore
+            bucket_permissions = bucket.get_permission(action, resource)
             if bucket_permissions == PermissionResult.DENIED:
                 return 403, {}, ""
 
@@ -1420,20 +1454,12 @@ class S3Response(BaseResponse):
 
         body = self.body or b""
 
-        if (
-            request.headers.get("x-amz-content-sha256", None)
-            == "STREAMING-AWS4-HMAC-SHA256-PAYLOAD"
-        ):
-            body = self._handle_v4_chunk_signatures(
-                body, int(request.headers["x-amz-decoded-content-length"])
-            )
-
         if method == "GET":
             return self._key_response_get(
                 bucket_name, query, key_name, headers=request.headers
             )
         elif method == "PUT":
-            return self._key_response_put(request, body, bucket_name, query, key_name)
+            return self._key_response_put(query, key_name)
         elif method == "HEAD":
             return self.head_object(
                 bucket_name, query, key_name, headers=request.headers
@@ -1588,275 +1614,81 @@ class S3Response(BaseResponse):
             not_modified = True
         return key, not_modified
 
-    def _key_response_put(
-        self,
-        request: Any,
-        body: bytes,
-        bucket_name: str,
-        query: Dict[str, Any],
-        key_name: str,
-    ) -> TYPE_RESPONSE:
+    def _key_response_put(self, query: Dict[str, Any], key_name: str) -> TYPE_RESPONSE:
         self._set_action("KEY", "PUT", query)
         self._authenticate_and_authorize_s3_action(
-            bucket_name=bucket_name, key_name=key_name
+            bucket_name=self.bucket_name, key_name=key_name
         )
 
-        response_headers = self._get_cors_headers_other(request.headers, bucket_name)
         if query.get("uploadId") and query.get("partNumber"):
-            upload_id = query["uploadId"][0]
-            part_number = int(query["partNumber"][0])
-            if "x-amz-copy-source" in request.headers:
-                copy_source = request.headers.get("x-amz-copy-source")
-                if isinstance(copy_source, bytes):
-                    copy_source = copy_source.decode("utf-8")
-                copy_source_parsed = urlparse(copy_source)
-                src_bucket, src_key = (
-                    unquote(copy_source_parsed.path).lstrip("/").split("/", 1)
-                )
-                src_version_id = parse_qs(copy_source_parsed.query).get(
-                    "versionId", [None]
-                )[0]
-                src_range = request.headers.get("x-amz-copy-source-range", "").split(
-                    "bytes="
-                )[-1]
-
-                try:
-                    start_byte, end_byte = src_range.split("-")
-                    start_byte, end_byte = int(start_byte), int(end_byte)
-                except ValueError:
-                    start_byte, end_byte = None, None
-
-                if self.backend.get_object(
-                    src_bucket, src_key, version_id=src_version_id
-                ):
-                    key = self.backend.upload_part_copy(
-                        bucket_name,
-                        upload_id,
-                        part_number,
-                        src_bucket_name=src_bucket,
-                        src_key_name=src_key,
-                        src_version_id=src_version_id,
-                        start_byte=start_byte,
-                        end_byte=end_byte,
-                    )
-                else:
-                    return 404, response_headers, ""
-
-                template = self.response_template(S3_MULTIPART_UPLOAD_RESPONSE)
-                response = template.render(part=key)
+            if "x-amz-copy-source" in self.headers:
+                return self.upload_part_copy()
             else:
-                if part_number > 10000:
-                    raise InvalidMaxPartNumberArgument(part_number)
-                key = self.backend.upload_part(
-                    bucket_name, upload_id, part_number, body
-                )
-                response = ""
-            response_headers.update(key.response_dict)
-            response_headers["content-length"] = str(len(response))
-            return 200, response_headers, response
-
-        storage_class = request.headers.get("x-amz-storage-class", "STANDARD")
-        encryption = request.headers.get("x-amz-server-side-encryption", None)
-        kms_key_id = request.headers.get(
-            "x-amz-server-side-encryption-aws-kms-key-id", None
-        )
-
-        checksum_algorithm = request.headers.get("x-amz-sdk-checksum-algorithm", "")
-        checksum_header = f"x-amz-checksum-{checksum_algorithm.lower()}"
-        checksum_value = request.headers.get(checksum_header)
-        if not checksum_value and checksum_algorithm:
-            # Extract the checksum-value from the body first
-            search = re.search(rb"x-amz-checksum-\w+:(.+={1,2})", body)
-            checksum_value = search.group(1) if search else None
-
-        if checksum_value:
-            # TODO: AWS computes the provided value and verifies it's the same
-            # Afterwards, it should be returned in every subsequent call
-            if isinstance(checksum_value, bytes):
-                checksum_value = checksum_value.decode("utf-8")
-            response_headers.update({checksum_header: checksum_value})
-        elif checksum_algorithm:
-            # If the value is not provided, we compute it and only return it as part of this request
-            checksum_value = compute_checksum(body, algorithm=checksum_algorithm)
-            response_headers.update({checksum_header: checksum_value})
-
-        # Extract the actual data from the body second
-        if (
-            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"])
-            )
-
-        bucket_key_enabled = request.headers.get(
-            "x-amz-server-side-encryption-bucket-key-enabled", None
-        )
-        if bucket_key_enabled is not None:
-            bucket_key_enabled = str(bucket_key_enabled).lower()
+                return self.upload_part()
 
-        bucket = self.backend.get_bucket(bucket_name)
+        bucket = self.backend.get_bucket(self.bucket_name)
         lock_enabled = bucket.object_lock_enabled
 
-        lock_mode = request.headers.get("x-amz-object-lock-mode", None)
-        lock_until = request.headers.get("x-amz-object-lock-retain-until-date", None)
-        legal_hold = request.headers.get("x-amz-object-lock-legal-hold", None)
+        lock_mode = self.headers.get("x-amz-object-lock-mode")
+        lock_until = self.headers.get("x-amz-object-lock-retain-until-date", None)
+        legal_hold = self.headers.get("x-amz-object-lock-legal-hold")
 
         if lock_mode or lock_until or legal_hold == "ON":
-            if not request.headers.get("Content-Md5"):
+            if not self.headers.get("Content-Md5"):
                 raise InvalidContentMD5
             if not lock_enabled:
                 raise LockNotEnabled
 
         elif lock_enabled and bucket.has_default_lock:
-            if not request.headers.get("Content-Md5"):
+            if not self.headers.get("Content-Md5"):
                 raise InvalidContentMD5
-            lock_until = bucket.default_retention()
-            lock_mode = bucket.default_lock_mode
-
-        acl = self._acl_from_headers(request.headers)
-        if acl is None:
-            acl = bucket.acl
-        tagging = self._tagging_from_headers(request.headers)
-
-        if "versionId" in query:
-            version_id = query["versionId"][0]
-        else:
-            version_id = None
 
         if "retention" in query:
-            if not lock_enabled:
-                raise LockNotEnabled
-            retention = self._mode_until_from_body()
-            self.backend.put_object_retention(
-                bucket_name, key_name, version_id=version_id, retention=retention
-            )
-            return 200, response_headers, ""
+            return self.put_object_retention()
 
         if "legal-hold" in query:
-            if not lock_enabled:
-                raise LockNotEnabled
-            legal_hold_status = self._legal_hold_status_from_xml(body)
-            self.backend.put_object_legal_hold(
-                bucket_name, key_name, version_id, legal_hold_status
-            )
-            return 200, response_headers, ""
+            return self.put_object_legal_hold()
 
         if "acl" in query:
-            self.backend.put_object_acl(bucket_name, key_name, acl)
-            return 200, response_headers, ""
+            return self.put_object_acl()
 
         if "tagging" in query:
-            key_to_tag = self.backend.get_object(
-                bucket_name, key_name, version_id=version_id
-            )
-            tagging = self._tagging_from_xml(body)
-            self.backend.put_object_tagging(key_to_tag, tagging, key_name)
-            return 200, response_headers, ""
+            return self.put_object_tagging()
 
-        if "x-amz-copy-source" in request.headers:
-            # Copy key
-            # you can have a quoted ?version=abc with a version Id, so work on
-            # we need to parse the unquoted string first
-            copy_source = request.headers.get("x-amz-copy-source")
-            if isinstance(copy_source, bytes):
-                copy_source = copy_source.decode("utf-8")
-            copy_source_parsed = urlparse(copy_source)
-            src_bucket, src_key = (
-                unquote(copy_source_parsed.path).lstrip("/").split("/", 1)
-            )
-            src_version_id = parse_qs(copy_source_parsed.query).get(
-                "versionId", [None]
-            )[0]
+        if "x-amz-copy-source" in self.headers:
+            return self.copy_object()
 
-            key_to_copy = self.backend.get_object(
-                src_bucket, src_key, version_id=src_version_id
-            )
+        return self.put_object()
 
-            if key_to_copy is not None:
-                if "x-amz-copy-source-if-none-match" in request.headers:
-                    requested_etag = request.headers["x-amz-copy-source-if-none-match"]
-                    if requested_etag in [key_to_copy.etag, key_to_copy.etag[1:-1]]:
-                        raise PreconditionFailed(
-                            failed_condition="x-amz-copy-source-If-None-Match"
-                        )
+    def put_object(self) -> TYPE_RESPONSE:
+        key_name = self.parse_key_name()
+        response_headers = self._get_cors_headers_other(self.headers, self.bucket_name)
 
-                if key_to_copy.storage_class in ARCHIVE_STORAGE_CLASSES:
-                    if (
-                        key_to_copy.response_dict.get("x-amz-restore") is None
-                        or 'ongoing-request="true"'
-                        in key_to_copy.response_dict.get(  # type: ignore
-                            "x-amz-restore"
-                        )
-                    ):
-                        raise ObjectNotInActiveTierError(key_to_copy)
+        storage_class = self.headers.get("x-amz-storage-class", "STANDARD")
+        encryption = self.headers.get("x-amz-server-side-encryption")
+        kms_key_id = self.headers.get("x-amz-server-side-encryption-aws-kms-key-id")
+        bucket_key_enabled = self.headers.get(
+            "x-amz-server-side-encryption-bucket-key-enabled"
+        )
+        if bucket_key_enabled is not None:
+            bucket_key_enabled = str(bucket_key_enabled).lower()
 
-                website_redirect_location = request.headers.get(
-                    "x-amz-website-redirect-location"
-                )
+        checksum_algorithm, checksum_value = self._get_checksum(response_headers)
 
-                mdirective = request.headers.get("x-amz-metadata-directive")
-                metadata = metadata_from_headers(request.headers)
-                self.backend.copy_object(
-                    key_to_copy,
-                    bucket_name,
-                    key_name,
-                    storage=request.headers.get("x-amz-storage-class"),
-                    kms_key_id=kms_key_id,
-                    encryption=encryption,
-                    bucket_key_enabled=bucket_key_enabled,
-                    mdirective=mdirective,
-                    metadata=metadata,
-                    website_redirect_location=website_redirect_location,
-                    lock_mode=lock_mode,
-                    lock_legal_status=legal_hold,
-                    lock_until=lock_until,
-                    provided_version_id=src_version_id,
-                )
-            else:
-                if src_version_id:
-                    raise MissingVersion()
-                raise MissingKey(key=src_key)
-
-            new_key: FakeKey = self.backend.get_object(bucket_name, key_name)  # type: ignore
-
-            if acl is not None:
-                new_key.set_acl(acl)
-
-            tdirective = request.headers.get("x-amz-tagging-directive")
-            if tdirective == "REPLACE":
-                tagging = self._tagging_from_headers(request.headers)
-                self.backend.put_object_tagging(new_key, tagging)
-            if key_to_copy.version_id != "null":
-                response_headers["x-amz-copy-source-version-id"] = (
-                    key_to_copy.version_id
-                )
+        bucket = self.backend.get_bucket(self.bucket_name)
+        lock_enabled = bucket.object_lock_enabled
 
-            # checksum stuff, do we need to compute hash of the copied object
-            checksum_algorithm = request.headers.get("x-amz-checksum-algorithm")
-            if checksum_algorithm:
-                checksum_value = compute_checksum(
-                    new_key.value, algorithm=checksum_algorithm
-                ).decode("utf-8")
-                response_headers.update(
-                    {"Checksum": {f"Checksum{checksum_algorithm}": checksum_value}}
-                )
-                # By default, the checksum-details for the copy will be the same as the original
-                # But if another algorithm is provided during the copy-operation, we override the values
-                new_key.checksum_algorithm = checksum_algorithm
-                new_key.checksum_value = checksum_value
-
-            template = self.response_template(S3_OBJECT_COPY_RESPONSE)
-            response_headers.update(new_key.response_dict)
-            response = template.render(key=new_key)
-            response_headers["content-length"] = str(len(response))
-            return 200, response_headers, response
+        legal_hold, lock_mode, lock_until = self._get_lock_details(bucket, lock_enabled)
+
+        acl = self._acl_from_headers(self.headers)
+        if acl is None:
+            acl = bucket.acl
+        tagging = self._tagging_from_headers(self.headers)
 
-        # Initial data
         new_key = self.backend.put_object(
-            bucket_name,
+            self.bucket_name,
             key_name,
-            body,
+            self.body or b"",
             storage=storage_class,
             encryption=encryption,
             kms_key_id=kms_key_id,
@@ -1866,23 +1698,279 @@ class S3Response(BaseResponse):
             lock_until=lock_until,
             checksum_value=checksum_value,
         )
-
-        metadata = metadata_from_headers(request.headers)
-        metadata.update(metadata_from_headers(query))
+        metadata = metadata_from_headers(self.headers)
+        metadata.update(metadata_from_headers(self.querystring))
         new_key.set_metadata(metadata)
         new_key.set_acl(acl)
-        new_key.website_redirect_location = request.headers.get(
+        new_key.website_redirect_location = self.headers.get(
             "x-amz-website-redirect-location"
         )
         if checksum_algorithm:
             new_key.checksum_algorithm = checksum_algorithm
         self.backend.put_object_tagging(new_key, tagging)
-
         response_headers.update(new_key.response_dict)
         # Remove content-length - the response body is empty for this request
         response_headers.pop("content-length", None)
         return 200, response_headers, ""
 
+    def _get_checksum(
+        self, response_headers: Dict[str, Any]
+    ) -> Tuple[str, Optional[str]]:
+        checksum_algorithm = self.headers.get("x-amz-sdk-checksum-algorithm", "")
+        checksum_header = f"x-amz-checksum-{checksum_algorithm.lower()}"
+        checksum_value = self.headers.get(checksum_header)
+        if not checksum_value and checksum_algorithm:
+            # Extract the checksum-value from the body first
+            search = re.search(rb"x-amz-checksum-\w+:(.+={1,2})", self.raw_body)
+            checksum_value = search.group(1) if search else None
+        if checksum_value:
+            # TODO: AWS computes the provided value and verifies it's the same
+            # Afterwards, it should be returned in every subsequent call
+            if isinstance(checksum_value, bytes):
+                checksum_value = checksum_value.decode("utf-8")
+            response_headers.update({checksum_header: checksum_value})
+        elif checksum_algorithm:
+            # If the value is not provided, we compute it and only return it as part of this request
+            checksum_value = compute_checksum(
+                self.raw_body, algorithm=checksum_algorithm
+            )
+            response_headers.update({checksum_header: checksum_value})
+        return checksum_algorithm, checksum_value
+
+    def copy_object(self) -> TYPE_RESPONSE:
+        # you can have a quoted ?version=abc with a version Id, so work on
+        # we need to parse the unquoted string first
+        copy_source = self.headers.get("x-amz-copy-source")
+        if isinstance(copy_source, bytes):
+            copy_source = copy_source.decode("utf-8")
+        copy_source_parsed = urlparse(copy_source)
+        src_bucket, src_key = unquote(copy_source_parsed.path).lstrip("/").split("/", 1)
+        src_version_id = parse_qs(copy_source_parsed.query).get("versionId", [None])[0]
+
+        key_to_copy = self.backend.get_object(
+            src_bucket, src_key, version_id=src_version_id
+        )
+        key_name = self.parse_key_name()
+
+        bucket = self.backend.get_bucket(self.bucket_name)
+        lock_enabled = bucket.object_lock_enabled
+
+        encryption = self.headers.get("x-amz-server-side-encryption")
+        kms_key_id = self.headers.get("x-amz-server-side-encryption-aws-kms-key-id")
+        bucket_key_enabled = self.headers.get(
+            "x-amz-server-side-encryption-bucket-key-enabled"
+        )
+        if bucket_key_enabled is not None:
+            bucket_key_enabled = str(bucket_key_enabled).lower()
+
+        legal_hold, lock_mode, lock_until = self._get_lock_details(bucket, lock_enabled)
+
+        if key_to_copy is not None:
+            if "x-amz-copy-source-if-none-match" in self.headers:
+                requested_etag = self.headers["x-amz-copy-source-if-none-match"]
+                if requested_etag in [key_to_copy.etag, key_to_copy.etag[1:-1]]:
+                    raise PreconditionFailed(
+                        failed_condition="x-amz-copy-source-If-None-Match"
+                    )
+
+            if key_to_copy.storage_class in ARCHIVE_STORAGE_CLASSES:
+                if key_to_copy.response_dict.get(
+                    "x-amz-restore"
+                ) is None or 'ongoing-request="true"' in key_to_copy.response_dict.get(  # type: ignore
+                    "x-amz-restore"
+                ):
+                    raise ObjectNotInActiveTierError(key_to_copy)
+
+            website_redirect_location = self.headers.get(
+                "x-amz-website-redirect-location"
+            )
+
+            mdirective = self.headers.get("x-amz-metadata-directive")
+            metadata = metadata_from_headers(self.headers)
+            self.backend.copy_object(
+                key_to_copy,
+                self.bucket_name,
+                key_name,
+                storage=self.headers.get("x-amz-storage-class"),
+                kms_key_id=kms_key_id,
+                encryption=encryption,
+                bucket_key_enabled=bucket_key_enabled,
+                mdirective=mdirective,
+                metadata=metadata,
+                website_redirect_location=website_redirect_location,
+                lock_mode=lock_mode,
+                lock_legal_status=legal_hold,
+                lock_until=lock_until,
+                provided_version_id=src_version_id,
+            )
+        else:
+            if src_version_id:
+                raise MissingVersion()
+            raise MissingKey(key=src_key)
+
+        new_key: FakeKey = self.backend.get_object(self.bucket_name, key_name)  # type: ignore
+
+        acl = self._acl_from_headers(self.headers)
+        if acl is None:
+            acl = bucket.acl
+        if acl is not None:
+            new_key.set_acl(acl)
+
+        response_headers = self._get_cors_headers_other(self.headers, self.bucket_name)
+        tdirective = self.headers.get("x-amz-tagging-directive")
+        if tdirective == "REPLACE":
+            tagging = self._tagging_from_headers(self.headers)
+            self.backend.put_object_tagging(new_key, tagging)
+        if key_to_copy.version_id != "null":
+            response_headers["x-amz-copy-source-version-id"] = key_to_copy.version_id
+
+        # checksum stuff, do we need to compute hash of the copied object
+        checksum_algorithm = self.headers.get("x-amz-checksum-algorithm")
+        if checksum_algorithm:
+            checksum_value = compute_checksum(
+                new_key.value, algorithm=checksum_algorithm
+            ).decode("utf-8")
+            response_headers.update(
+                {"Checksum": {f"Checksum{checksum_algorithm}": checksum_value}}
+            )
+            # By default, the checksum-details for the copy will be the same as the original
+            # But if another algorithm is provided during the copy-operation, we override the values
+            new_key.checksum_algorithm = checksum_algorithm
+            new_key.checksum_value = checksum_value
+
+        template = self.response_template(S3_OBJECT_COPY_RESPONSE)
+        response_headers.update(new_key.response_dict)
+        response = template.render(key=new_key)
+        response_headers["content-length"] = str(len(response))
+        return 200, response_headers, response
+
+    def _get_lock_details(
+        self, bucket: "FakeBucket", lock_enabled: bool
+    ) -> Tuple[Optional[str], Optional[str], Optional[str]]:
+        lock_mode = self.headers.get("x-amz-object-lock-mode")
+        lock_until = self.headers.get("x-amz-object-lock-retain-until-date")
+        legal_hold = self.headers.get("x-amz-object-lock-legal-hold")
+        if lock_mode or lock_until or legal_hold == "ON":
+            pass
+        elif lock_enabled and bucket.has_default_lock:
+            lock_until = bucket.default_retention()
+            lock_mode = bucket.default_lock_mode
+        return legal_hold, lock_mode, lock_until
+
+    def put_object_tagging(self) -> TYPE_RESPONSE:
+        key_name = self.parse_key_name()
+        version_id = self._get_param("versionId")
+        key_to_tag = self.backend.get_object(
+            self.bucket_name, key_name, version_id=version_id
+        )
+        tagging = self._tagging_from_xml()
+        self.backend.put_object_tagging(key_to_tag, tagging, key_name)
+
+        response_headers = self._get_cors_headers_other(self.headers, self.bucket_name)
+        self._get_checksum(response_headers)
+        return 200, response_headers, ""
+
+    def put_object_acl(self) -> TYPE_RESPONSE:
+        acl = self._acl_from_headers(self.headers)
+        if acl is None:
+            bucket = self.backend.get_bucket(self.bucket_name)
+            acl = bucket.acl
+        key_name = self.parse_key_name()
+        self.backend.put_object_acl(self.bucket_name, key_name, acl)
+
+        response_headers = self._get_cors_headers_other(self.headers, self.bucket_name)
+        self._get_checksum(response_headers)
+        return 200, response_headers, ""
+
+    def put_object_legal_hold(self) -> TYPE_RESPONSE:
+        version_id = self._get_param("versionId")
+        bucket = self.backend.get_bucket(self.bucket_name)
+        lock_enabled = bucket.object_lock_enabled
+        key_name = self.parse_key_name()
+
+        if not lock_enabled:
+            raise LockNotEnabled
+        legal_hold_status = self._legal_hold_status_from_xml(self.body)
+        self.backend.put_object_legal_hold(
+            self.bucket_name, key_name, version_id, legal_hold_status
+        )
+
+        response_headers = self._get_cors_headers_other(self.headers, self.bucket_name)
+        self._get_checksum(response_headers)
+        return 200, response_headers, ""
+
+    def put_object_retention(self) -> TYPE_RESPONSE:
+        version_id = self._get_param("versionId")
+        bucket = self.backend.get_bucket(self.bucket_name)
+        lock_enabled = bucket.object_lock_enabled
+        key_name = self.parse_key_name()
+
+        if not lock_enabled:
+            raise LockNotEnabled
+        retention = self._mode_until_from_body()
+        self.backend.put_object_retention(
+            self.bucket_name, key_name, version_id=version_id, retention=retention
+        )
+
+        response_headers = self._get_cors_headers_other(self.headers, self.bucket_name)
+        self._get_checksum(response_headers)
+        return 200, response_headers, ""
+
+    def upload_part(self) -> TYPE_RESPONSE:
+        upload_id = self._get_param("uploadId")
+        part_number = self._get_int_param("partNumber")
+
+        if part_number > 10000:
+            raise InvalidMaxPartNumberArgument(part_number)
+        key = self.backend.upload_part(
+            self.bucket_name, upload_id, part_number, self.body
+        )
+
+        response_headers = self._get_cors_headers_other(self.headers, self.bucket_name)
+        self._get_checksum(response_headers)
+        response_headers.update(key.response_dict)
+        response_headers["content-length"] = "0"
+        return 200, response_headers, ""
+
+    def upload_part_copy(self) -> TYPE_RESPONSE:
+        response_headers = self._get_cors_headers_other(self.headers, self.bucket_name)
+        self._get_checksum(response_headers)
+
+        upload_id = self._get_param("uploadId")
+        part_number = self._get_int_param("partNumber")
+
+        copy_source = self.headers.get("x-amz-copy-source")
+        if isinstance(copy_source, bytes):
+            copy_source = copy_source.decode("utf-8")
+        copy_source_parsed = urlparse(copy_source)
+        src_bucket, src_key = unquote(copy_source_parsed.path).lstrip("/").split("/", 1)
+        src_version_id = parse_qs(copy_source_parsed.query).get("versionId", [None])[0]
+        src_range = self.headers.get("x-amz-copy-source-range", "").split("bytes=")[-1]
+        try:
+            start_byte, end_byte = src_range.split("-")
+            start_byte, end_byte = int(start_byte), int(end_byte)
+        except ValueError:
+            start_byte, end_byte = None, None
+        if self.backend.get_object(src_bucket, src_key, version_id=src_version_id):
+            key = self.backend.upload_part_copy(
+                self.bucket_name,
+                upload_id,
+                part_number,
+                src_bucket_name=src_bucket,
+                src_key_name=src_key,
+                src_version_id=src_version_id,
+                start_byte=start_byte,
+                end_byte=end_byte,
+            )
+        else:
+            return 404, response_headers, ""
+        template = self.response_template(S3_MULTIPART_UPLOAD_RESPONSE)
+        response = template.render(part=key)
+
+        response_headers.update(key.response_dict)
+        response_headers["content-length"] = str(len(response))
+        return 200, response_headers, response
+
     def head_object(
         self,
         bucket_name: str,
@@ -2110,8 +2198,8 @@ class S3Response(BaseResponse):
                 tags[tag[0]] = tag[1][0]
         return tags
 
-    def _tagging_from_xml(self, xml: bytes) -> Dict[str, str]:
-        parsed_xml = xmltodict.parse(xml, force_list={"Tag": True})
+    def _tagging_from_xml(self) -> Dict[str, str]:
+        parsed_xml = xmltodict.parse(self.body, force_list={"Tag": True})
 
         tags = {}
         for tag in parsed_xml["Tagging"]["TagSet"]["Tag"]:
diff --git a/moto/transfer/types.py b/moto/transfer/types.py
index c02527092..efaef0997 100644
--- a/moto/transfer/types.py
+++ b/moto/transfer/types.py
@@ -37,9 +37,7 @@ class User(BaseModel):
         if self.arn == "":
             self.arn = f"arn:aws:transfer:{self.user_name}:{datetime.now().strftime('%Y%m%d%H%M%S')}"
 
-    def to_dict(
-        self,
-    ) -> Dict[str, Any]:  # type: ignore[misc]
+    def to_dict(self) -> Dict[str, Any]:
         user = {
             "HomeDirectory": self.home_directory,
             "HomeDirectoryType": self.home_directory_type,
@@ -152,7 +150,7 @@ class Server(BaseModel):
         if self.as2_service_managed_egress_ip_addresses == []:
             self.as2_service_managed_egress_ip_addresses.append("0.0.0.0/0")
 
-    def to_dict(self) -> Dict[str, Any]:  # type: ignore[misc]
+    def to_dict(self) -> Dict[str, Any]:
         on_upload = []
         on_partial_upload = []
         if self.workflow_details is not 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_acmpca/test_acmpca.py b/tests/test_acmpca/test_acmpca.py
index a9c592cd5..56b9c0df6 100644
--- a/tests/test_acmpca/test_acmpca.py
+++ b/tests/test_acmpca/test_acmpca.py
@@ -1,4 +1,5 @@
 import datetime
+import pickle
 
 import boto3
 import cryptography.hazmat.primitives.asymmetric.rsa
@@ -388,8 +389,16 @@ def test_import_certificate_authority_certificate():
         CertificateAuthorityType="SUBORDINATE",
     )["CertificateAuthorityArn"]
 
-    cert = create_cert()
+    with pytest.raises(ClientError) as exc:
+        client.import_certificate_authority_certificate(
+            CertificateAuthorityArn=ca_arn,
+            Certificate="invalid certificate pem",
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "MalformedCertificateAuthorityException"
+    assert err["Message"] == "Malformed certificate."
 
+    cert = create_cert()
     client.import_certificate_authority_certificate(
         CertificateAuthorityArn=ca_arn,
         Certificate=cert,
@@ -513,6 +522,46 @@ def test_untag_certificate_authority():
     assert resp["Tags"] == [{"Key": "t2", "Value": "v2"}]
 
 
+@mock_aws
+def test_acmpca_backend_is_serialisable():
+    # This test ensures that the backend does not store any non-picklable objects (e.g. native OpenSSL objects)
+    # This is important for downstream projects like LocalStack
+
+    client = boto3.client("acm-pca", region_name="eu-west-1")
+
+    ca_arn = client.create_certificate_authority(
+        CertificateAuthorityConfiguration={
+            "KeyAlgorithm": "RSA_4096",
+            "SigningAlgorithm": "SHA512WITHRSA",
+            "Subject": {"CommonName": "yscb41lw.test"},
+        },
+        CertificateAuthorityType="SUBORDINATE",
+    )["CertificateAuthorityArn"]
+
+    cert = create_cert()
+    client.import_certificate_authority_certificate(
+        CertificateAuthorityArn=ca_arn,
+        Certificate=cert,
+    )
+    client.update_certificate_authority(
+        CertificateAuthorityArn=ca_arn,
+        Status="DISABLED",
+    )
+
+    csr = client.get_certificate_authority_csr(CertificateAuthorityArn=ca_arn)["Csr"]
+    client.issue_certificate(
+        CertificateAuthorityArn=ca_arn,
+        Csr=csr,
+        SigningAlgorithm="SHA512WITHRSA",
+        Validity={"Type": "YEARS", "Value": 10},
+    )
+    client.describe_certificate_authority(CertificateAuthorityArn=ca_arn)
+
+    from moto.acmpca import acmpca_backends
+
+    assert pickle.dumps(acmpca_backends)
+
+
 def create_csr(private_key, country, state, org, cn):
     csr = (
         cryptography.x509.CertificateSigningRequestBuilder()
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"