Repository: getmoto/moto
Pull request #6309 (2023-05-10 10:44:48)

S3: fix checksum calculation
While working against AWS, I realised moto was returning different checksum after copying an object and recalculating its checksum. It seems the calculation was bit wrong, as it was using `hexdigest` instead of the binary `digest` when passing the hash to the base64 encoding.
I've tried to add a test to cover all checksum calculation by using a checksum recalculation property of `CopyObject`.

Also fixed CopyObject behaviour, when copying a key that has a Checksum, if not asked to be recalculated, it will also copy it to the destination key. This has been verified with AWS (same with the values, so we can check that the recalculation works properly, even when not provided by boto in the first request like in `test_copy_key_boto3_with_both_sha256_checksum`). 

bblommers: Hey @bentsku, the linter has been fixed now - if you merge in the latest master, the build should pass again

Patch:
diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py
index 6a93607d0..4b3373fb7 100644
--- a/moto/ec2/exceptions.py
+++ b/moto/ec2/exceptions.py
@@ -234,6 +234,15 @@ class RouteAlreadyExistsError(EC2ClientError):
         )
 
 
+class RouteNotSupportedError(EC2ClientError):
+    def __init__(self, vpce_id: str):
+        super().__init__(
+            "RouteNotSupported",
+            f"Route table contains unsupported route target: {vpce_id}. "
+            f"VPC Endpoints of this type cannot be used as route targets.",
+        )
+
+
 class InvalidInstanceIdError(EC2ClientError):
     def __init__(self, instance_id: Any):
         if isinstance(instance_id, str):
diff --git a/moto/ec2/models/route_tables.py b/moto/ec2/models/route_tables.py
index 4215a320c..de6217d09 100644
--- a/moto/ec2/models/route_tables.py
+++ b/moto/ec2/models/route_tables.py
@@ -19,6 +19,7 @@ from ..exceptions import (
     InvalidAssociationIdError,
     InvalidDestinationCIDRBlockParameterError,
     RouteAlreadyExistsError,
+    RouteNotSupportedError,
 )
 from ..utils import (
     EC2_RESOURCE_TO_PREFIX,
@@ -141,6 +142,7 @@ class Route(CloudFormationModel):
         interface: Optional[NetworkInterface] = None,
         vpc_pcx: Optional[VPCPeeringConnection] = None,
         carrier_gateway: Optional[CarrierGateway] = None,
+        vpc_endpoint_id: Optional[str] = None,
     ):
         self.id = generate_route_id(
             route_table.id,
@@ -161,6 +163,7 @@ class Route(CloudFormationModel):
         self.interface = interface
         self.vpc_pcx = vpc_pcx
         self.carrier_gateway = carrier_gateway
+        self.vpc_endpoint_id = vpc_endpoint_id
 
     @property
     def physical_resource_id(self) -> str:
@@ -359,6 +362,7 @@ class RouteBackend:
         interface_id: Optional[str] = None,
         vpc_peering_connection_id: Optional[str] = None,
         carrier_gateway_id: Optional[str] = None,
+        vpc_endpoint_id: Optional[str] = None,
     ) -> Route:
         gateway = None
         nat_gateway = None
@@ -368,6 +372,13 @@ class RouteBackend:
         destination_prefix_list = None
         carrier_gateway = None
 
+        if vpc_endpoint_id:
+            vpce = self.describe_vpc_endpoints(vpc_end_point_ids=[vpc_endpoint_id])  # type: ignore[attr-defined]
+            if not vpce[0].endpoint_type == "GatewayLoadBalancer":
+                # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/client/create_route.html
+                # VpcEndpointId (string) – The ID of a VPC endpoint. Supported for Gateway Load Balancer endpoints only.
+                raise RouteNotSupportedError(vpc_endpoint_id)
+
         route_table = self.get_route_table(route_table_id)
 
         if interface_id:
@@ -414,6 +425,7 @@ class RouteBackend:
             transit_gateway=transit_gateway,
             interface=interface,
             carrier_gateway=carrier_gateway,
+            vpc_endpoint_id=vpc_endpoint_id,
             vpc_pcx=self.get_vpc_peering_connection(vpc_peering_connection_id)  # type: ignore[attr-defined]
             if vpc_peering_connection_id
             else None,
diff --git a/moto/ec2/responses/route_tables.py b/moto/ec2/responses/route_tables.py
index bab437680..be8aeaf04 100644
--- a/moto/ec2/responses/route_tables.py
+++ b/moto/ec2/responses/route_tables.py
@@ -25,6 +25,7 @@ class RouteTables(EC2BaseResponse):
         interface_id = self._get_param("NetworkInterfaceId")
         pcx_id = self._get_param("VpcPeeringConnectionId")
         carrier_gateway_id = self._get_param("CarrierGatewayId")
+        vpc_endpoint_id = self._get_param("VpcEndpointId")
 
         self.ec2_backend.create_route(
             route_table_id,
@@ -39,6 +40,7 @@ class RouteTables(EC2BaseResponse):
             interface_id=interface_id,
             vpc_peering_connection_id=pcx_id,
             carrier_gateway_id=carrier_gateway_id,
+            vpc_endpoint_id=vpc_endpoint_id,
         )
 
         template = self.response_template(CREATE_ROUTE_RESPONSE)
@@ -212,6 +214,11 @@ DESCRIBE_ROUTE_TABLES_RESPONSE = """
                   <origin>CreateRoute</origin>
                   <state>active</state>
                 {% endif %}
+                {% if route.vpc_endpoint_id %}
+                  <gatewayId>{{ route.vpc_endpoint_id }}</gatewayId>
+                  <origin>CreateRoute</origin>
+                  <state>active</state>
+                {% endif %}
                 {% if route.instance %}
                   <instanceId>{{ route.instance.id }}</instanceId>
                   <origin>CreateRoute</origin>
diff --git a/moto/s3/models.py b/moto/s3/models.py
index 75ad72fac..c7f4e9527 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -2418,6 +2418,10 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
             # Object copied from Glacier object should not have expiry
             new_key.set_expiry(None)
 
+        if src_key.checksum_value:
+            new_key.checksum_value = src_key.checksum_value
+            new_key.checksum_algorithm = src_key.checksum_algorithm
+
         # Send notifications that an object was copied
         notifications.send_event(
             self.account_id, notifications.S3_OBJECT_CREATE_COPY, bucket, new_key
diff --git a/moto/s3/utils.py b/moto/s3/utils.py
index 855ce4a4f..fafabdcd1 100644
--- a/moto/s3/utils.py
+++ b/moto/s3/utils.py
@@ -200,7 +200,7 @@ def compute_checksum(body: bytes, algorithm: str) -> bytes:
     if algorithm == "SHA1":
         hashed_body = _hash(hashlib.sha1, (body,))
     elif algorithm == "CRC32" or algorithm == "CRC32C":
-        hashed_body = f"{binascii.crc32(body)}".encode("utf-8")
+        hashed_body = binascii.crc32(body).to_bytes(4, "big")
     else:
         hashed_body = _hash(hashlib.sha256, (body,))
     return base64.b64encode(hashed_body)
@@ -208,10 +208,10 @@ def compute_checksum(body: bytes, algorithm: str) -> bytes:
 
 def _hash(fn: Any, args: Any) -> bytes:
     try:
-        return fn(*args, usedforsecurity=False).hexdigest().encode("utf-8")
+        return fn(*args, usedforsecurity=False).digest()
     except TypeError:
         # The usedforsecurity-parameter is only available as of Python 3.9
-        return fn(*args).hexdigest().encode("utf-8")
+        return fn(*args).digest()
 
 
 def cors_matches_origin(origin_header: str, allowed_origins: List[str]) -> bool:
diff --git a/moto/secretsmanager/models.py b/moto/secretsmanager/models.py
index bb492c4ab..53d049ef0 100644
--- a/moto/secretsmanager/models.py
+++ b/moto/secretsmanager/models.py
@@ -88,6 +88,8 @@ class FakeSecret:
         self.auto_rotate_after_days = 0
         self.deleted_date: Optional[float] = None
         self.policy: Optional[str] = None
+        self.next_rotation_date: Optional[int] = None
+        self.last_rotation_date: Optional[int] = None
 
     def update(
         self,
@@ -167,9 +169,10 @@ class FakeSecret:
             "RotationEnabled": self.rotation_enabled,
             "RotationLambdaARN": self.rotation_lambda_arn,
             "RotationRules": {"AutomaticallyAfterDays": self.auto_rotate_after_days},
-            "LastRotatedDate": None,
+            "LastRotatedDate": self.last_rotation_date,
             "LastChangedDate": self.last_changed_date,
             "LastAccessedDate": None,
+            "NextRotationDate": self.next_rotation_date,
             "DeletedDate": self.deleted_date,
             "Tags": self.tags,
             "VersionIdsToStages": version_id_to_stages,
@@ -535,6 +538,10 @@ class SecretsManagerBackend(BaseBackend):
                     )
                     raise InvalidParameterException(msg)
 
+                self.secrets[secret_id].next_rotation_date = int(time.time()) + (
+                    int(rotation_period) * 86400
+                )
+
         secret = self.secrets[secret_id]
 
         # The rotation function must end with the versions of the secret in
@@ -625,7 +632,7 @@ class SecretsManagerBackend(BaseBackend):
                 secret.versions[new_version_id], new_version_id
             )
             secret.versions[new_version_id]["version_stages"] = ["AWSCURRENT"]
-
+        self.secrets[secret_id].last_rotation_date = int(time.time())
         return secret.to_short_dict()
 
     def get_random_password(


Test patch:
diff --git a/tests/test_ec2/test_route_tables.py b/tests/test_ec2/test_route_tables.py
index 51c37a646..70b70bd67 100644
--- a/tests/test_ec2/test_route_tables.py
+++ b/tests/test_ec2/test_route_tables.py
@@ -1041,6 +1041,62 @@ def test_create_route_with_unknown_egress_only_igw():
     err["Message"].should.equal("The eigw ID 'eoigw' does not exist")
 
 
+@mock_ec2
+def test_create_route_with_vpc_endpoint():
+    # Setup
+    _, ec2_client, route_table, vpc = setup_vpc()
+    dest_cidr = "0.0.0.0/0"
+    vpc_end_point = ec2_client.create_vpc_endpoint(
+        VpcId=vpc.id,
+        ServiceName="com.amazonaws.vpce.eu-central-1.vpce-svc-084fa044c50cb1290",
+        RouteTableIds=[route_table.id],
+        VpcEndpointType="GatewayLoadBalancer",
+    )
+    vpce_id = vpc_end_point["VpcEndpoint"]["VpcEndpointId"]
+
+    # Execute
+    ec2_client.create_route(
+        DestinationCidrBlock=dest_cidr,
+        VpcEndpointId=vpce_id,
+        RouteTableId=route_table.id,
+    )
+    rt = ec2_client.describe_route_tables()
+    new_route = rt["RouteTables"][-1]["Routes"][1]
+
+    # Verify
+    assert new_route["DestinationCidrBlock"] == dest_cidr
+    assert new_route["GatewayId"] == vpce_id
+
+
+@mock_ec2
+def test_create_route_with_invalid_vpc_endpoint():
+    # Setup
+    _, ec2_client, route_table, vpc = setup_vpc()
+    dest_cidr = "0.0.0.0/0"
+    vpc_end_point = ec2_client.create_vpc_endpoint(
+        VpcId=vpc.id,
+        ServiceName="com.amazonaws.us-east-1.s3",
+        RouteTableIds=[route_table.id],
+        VpcEndpointType="Gateway",
+    )
+    vpce_id = vpc_end_point["VpcEndpoint"]["VpcEndpointId"]
+
+    # Execute
+    with pytest.raises(ClientError) as ex:
+        ec2_client.create_route(
+            DestinationCidrBlock=dest_cidr,
+            VpcEndpointId=vpce_id,
+            RouteTableId=route_table.id,
+        )
+    # Verify
+    err = ex.value.response["Error"]
+    assert err["Code"] == "RouteNotSupported"
+    assert (
+        err["Message"] == f"Route table contains unsupported route target: {vpce_id}. "
+        "VPC Endpoints of this type cannot be used as route targets."
+    )
+
+
 @mock_ec2
 def test_associate_route_table_by_gateway():
     ec2 = boto3.client("ec2", region_name="us-west-1")
@@ -1087,3 +1143,17 @@ def test_associate_route_table_by_subnet():
     verify[0]["Associations"][0]["SubnetId"].should.equals(subnet_id)
     verify[0]["Associations"][0]["RouteTableAssociationId"].should.equal(assoc_id)
     verify[0]["Associations"][0].doesnt.have.key("GatewayId")
+
+
+def setup_vpc():
+    ec2_resource = boto3.resource("ec2", region_name="eu-central-1")
+    ec2_client = boto3.client("ec2", region_name="eu-central-1")
+
+    vpc = ec2_resource.create_vpc(CidrBlock="10.0.0.0/16")
+    ec2_resource.create_subnet(
+        VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-2a"
+    )
+
+    route_table = ec2_resource.create_route_table(VpcId=vpc.id)
+
+    return ec2_resource, ec2_client, route_table, vpc
diff --git a/tests/test_s3/test_s3_copyobject.py b/tests/test_s3/test_s3_copyobject.py
index 5a54a0537..701e8df58 100644
--- a/tests/test_s3/test_s3_copyobject.py
+++ b/tests/test_s3/test_s3_copyobject.py
@@ -46,7 +46,7 @@ def test_copy_key_boto3_with_sha256_checksum():
     key_name = "key"
     new_key = "new_key"
     bucket = "foobar"
-    expected_hash = "YWIzZDA3ZjMxNjljY2JkMGVkNmM0YjQ1ZGUyMTUxOWY5ZjkzOGM3MmQyNDEyNDk5OGFhYjk0OWNlODNiYjUxYg=="
+    expected_hash = "qz0H8xacy9DtbEtF3iFRn5+TjHLSQSSZiquUnOg7tRs="
 
     s3.create_bucket(Bucket=bucket)
     key = s3.Object("foobar", key_name)
@@ -730,3 +730,112 @@ def test_copy_object_in_place_with_bucket_encryption():
         Key=key,
     )
     assert response["ServerSideEncryption"] == "AES256"
+
+
+@mock_s3
+@pytest.mark.parametrize(
+    "algorithm",
+    ["CRC32", "SHA1", "SHA256"],
+)
+def test_copy_key_boto3_with_both_sha256_checksum(algorithm):
+    # This test will validate that moto S3 checksum calculations are correct
+    # We first create an object with a Checksum calculated by boto, by specifying ChecksumAlgorithm="SHA256"
+    # we then retrieve the right checksum from this request
+    # we copy the object while requesting moto to recalculate the checksum for that key
+    # we verify that both checksums are equal
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    source_key = "source-key"
+    dest_key = "dest-key"
+    bucket = "foobar"
+    body = b"checksum-test"
+    client.create_bucket(Bucket=bucket)
+
+    checksum_key = f"Checksum{algorithm}"
+
+    resp = client.put_object(
+        Bucket=bucket,
+        Key=source_key,
+        Body=body,
+        ChecksumAlgorithm=algorithm,
+    )
+    assert checksum_key in resp
+    checksum_by_boto = resp[checksum_key]
+
+    resp = client.copy_object(
+        Bucket=bucket,
+        CopySource=f"{bucket}/{source_key}",
+        Key=dest_key,
+        ChecksumAlgorithm=algorithm,
+    )
+
+    assert checksum_key in resp["CopyObjectResult"]
+    assert resp["CopyObjectResult"][checksum_key] == checksum_by_boto
+
+
+@mock_s3
+@pytest.mark.parametrize(
+    "algorithm, checksum",
+    [
+        ("CRC32", "lVk/nw=="),
+        ("SHA1", "jbXkHAsXUrubtL3dqDQ4w+7WXc0="),
+        ("SHA256", "1YQo81vx2VFUl0q5ccWISq8AkSBQQ0WO80S82TmfdIQ="),
+    ],
+)
+def test_copy_object_calculates_checksum(algorithm, checksum):
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    source_key = "source-key"
+    dest_key = "dest-key"
+    bucket = "foobar"
+    body = b"test-checksum"
+    client.create_bucket(Bucket=bucket)
+
+    checksum_key = f"Checksum{algorithm}"
+
+    resp = client.put_object(
+        Bucket=bucket,
+        Key=source_key,
+        Body=body,
+    )
+    assert checksum_key not in resp
+
+    resp = client.copy_object(
+        Bucket=bucket,
+        CopySource=f"{bucket}/{source_key}",
+        Key=dest_key,
+        ChecksumAlgorithm=algorithm,
+    )
+
+    assert checksum_key in resp["CopyObjectResult"]
+    assert resp["CopyObjectResult"][checksum_key] == checksum
+
+
+@mock_s3
+def test_copy_object_keeps_checksum():
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    source_key = "source-key"
+    dest_key = "dest-key"
+    bucket = "foobar"
+    body = b"test-checksum"
+    expected_checksum = "1YQo81vx2VFUl0q5ccWISq8AkSBQQ0WO80S82TmfdIQ="
+    client.create_bucket(Bucket=bucket)
+
+    # put an object with a checksum
+    resp = client.put_object(
+        Bucket=bucket,
+        Key=source_key,
+        Body=body,
+        ChecksumAlgorithm="SHA256",
+    )
+    assert "ChecksumSHA256" in resp
+    assert resp["ChecksumSHA256"] == expected_checksum
+
+    # do not specify the checksum
+    resp = client.copy_object(
+        Bucket=bucket,
+        CopySource=f"{bucket}/{source_key}",
+        Key=dest_key,
+    )
+
+    # assert that it kept the checksum from the source key
+    assert "ChecksumSHA256" in resp["CopyObjectResult"]
+    assert resp["CopyObjectResult"]["ChecksumSHA256"] == expected_checksum
diff --git a/tests/test_s3/test_s3_utils.py b/tests/test_s3/test_s3_utils.py
index a5bdd7344..5edd5a24b 100644
--- a/tests/test_s3/test_s3_utils.py
+++ b/tests/test_s3/test_s3_utils.py
@@ -124,24 +124,22 @@ def test_undo_clean_key_name(key, expected):
 
 
 def test_checksum_sha256():
-    checksum = b"ODdkMTQ5Y2I0MjRjMDM4NzY1NmYyMTFkMjU4OWZiNWIxZTE2MjI5OTIxMzA5ZTk4NTg4NDE5Y2NjYThhNzM2Mg=="
+    checksum = b"h9FJy0JMA4dlbyEdJYn7Wx4WIpkhMJ6YWIQZzMqKc2I="
     compute_checksum(b"somedata", "SHA256").should.equal(checksum)
     # Unknown algorithms fallback to SHA256 for now
     compute_checksum(b"somedata", algorithm="unknown").should.equal(checksum)
 
 
 def test_checksum_sha1():
-    compute_checksum(b"somedata", "SHA1").should.equal(
-        b"ZWZhYTMxMWFlNDQ4YTczNzRjMTIyMDYxYmZlZDk1MmQ5NDBlOWUzNw=="
-    )
+    compute_checksum(b"somedata", "SHA1").should.equal(b"76oxGuRIpzdMEiBhv+2VLZQOnjc=")
 
 
 def test_checksum_crc32():
-    compute_checksum(b"somedata", "CRC32").should.equal(b"MTM5MzM0Mzk1Mg==")
+    compute_checksum(b"somedata", "CRC32").should.equal(b"Uwy90A==")
 
 
 def test_checksum_crc32c():
-    compute_checksum(b"somedata", "CRC32C").should.equal(b"MTM5MzM0Mzk1Mg==")
+    compute_checksum(b"somedata", "CRC32C").should.equal(b"Uwy90A==")
 
 
 def test_cors_utils():
diff --git a/tests/test_secretsmanager/test_secretsmanager.py b/tests/test_secretsmanager/test_secretsmanager.py
index 6eee8b762..f2f78ec22 100644
--- a/tests/test_secretsmanager/test_secretsmanager.py
+++ b/tests/test_secretsmanager/test_secretsmanager.py
@@ -711,21 +711,41 @@ def test_cancel_rotate_secret():
 
 @mock_secretsmanager
 def test_rotate_secret():
-    conn = boto3.client("secretsmanager", region_name="us-west-2")
-    conn.create_secret(
-        Name=DEFAULT_SECRET_NAME, SecretString="foosecret", Description="foodescription"
-    )
-
-    rotated_secret = conn.rotate_secret(SecretId=DEFAULT_SECRET_NAME)
-
-    assert rotated_secret
-    assert rotated_secret["ARN"] != ""  # Test arn not empty
-    assert rotated_secret["Name"] == DEFAULT_SECRET_NAME
-    assert rotated_secret["VersionId"] != ""
-
-    describe_secret = conn.describe_secret(SecretId=DEFAULT_SECRET_NAME)
+    # Setup
+    frozen_time = datetime(2023, 5, 20, 10, 20, 30, tzinfo=tzlocal())
+    rotate_after_days = 10
+    with freeze_time(frozen_time):
+        conn = boto3.client("secretsmanager", region_name="us-west-2")
+        conn.create_secret(
+            Name=DEFAULT_SECRET_NAME,
+            SecretString="foosecret",
+            Description="foodescription",
+        )
 
-    assert describe_secret["Description"] == "foodescription"
+        # Execute
+        rotated_secret = conn.rotate_secret(
+            SecretId=DEFAULT_SECRET_NAME,
+            RotationRules={"AutomaticallyAfterDays": rotate_after_days},
+        )
+        describe_secret = conn.describe_secret(SecretId=DEFAULT_SECRET_NAME)
+
+        # Verify
+        assert rotated_secret
+        assert rotated_secret["ARN"] != ""  # Test arn not empty
+        assert rotated_secret["Name"] == DEFAULT_SECRET_NAME
+        assert rotated_secret["VersionId"] != ""
+        assert describe_secret["Description"] == "foodescription"
+        assert "NextRotationDate" in describe_secret
+        assert "LastRotatedDate" in describe_secret
+
+        # can't do freeze time tests in servermode tests
+        if settings.TEST_SERVER_MODE:
+            return
+
+        assert describe_secret["LastChangedDate"] == frozen_time
+        assert describe_secret["NextRotationDate"] == frozen_time + timedelta(
+            days=rotate_after_days
+        )
 
 
 @mock_secretsmanager