Route53: Intercept duplicate calls to change_rr_sets (#5725)

moto/route53/exceptions.py

@@ -162,3 +162,20 @@ class NoSuchDelegationSet(Route53ClientError):
     def __init__(self, delegation_set_id):
         super().__init__("NoSuchDelegationSet", delegation_set_id)
         self.content_type = "text/xml"
+
+
+class DnsNameInvalidForZone(Route53ClientError):
+    code = 400
+
+    def __init__(self, name, zone_name):
+        error_msg = (
+            f"""RRSet with DNS name {name} is not permitted in zone {zone_name}"""
+        )
+        super().__init__("InvalidChangeBatch", error_msg)
+
+
+class ChangeSetAlreadyExists(Route53ClientError):
+    code = 400
+
+    def __init__(self):
+        super().__init__("InvalidChangeBatch", "Provided Change is a duplicate")

moto/route53/models.py

@@ -1,4 +1,5 @@
 """Route53Backend class with methods for supported APIs."""
+import copy
 import itertools
 import re
 import string
@@ -17,6 +18,8 @@ from moto.route53.exceptions import (
     NoSuchQueryLoggingConfig,
     PublicZoneVPCAssociation,
     QueryLoggingConfigAlreadyExists,
+    DnsNameInvalidForZone,
+    ChangeSetAlreadyExists,
 )
 from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel
 from moto.moto_api._internal import mock_random as random
@@ -276,6 +279,7 @@ class FakeZone(CloudFormationModel):
         self.private_zone = private_zone
         self.rrsets = []
         self.delegation_set = delegation_set
+        self.rr_changes = []
 
     def add_rrset(self, record_set):
         record_set = RecordSet(record_set)
@@ -525,9 +529,14 @@ class Route53Backend(BaseBackend):
         is_truncated = next_record is not None
         return records, next_start_name, next_start_type, is_truncated
 
-    def change_resource_record_sets(self, zoneid, change_list):
+    def change_resource_record_sets(self, zoneid, change_list) -> None:
         the_zone = self.get_hosted_zone(zoneid)
+
+        if any([rr for rr in change_list if rr in the_zone.rr_changes]):
+            raise ChangeSetAlreadyExists
+
         for value in change_list:
+            original_change = copy.deepcopy(value)
             action = value["Action"]
 
             if action not in ("CREATE", "UPSERT", "DELETE"):
@@ -539,11 +548,9 @@ class Route53Backend(BaseBackend):
             cleaned_hosted_zone_name = the_zone.name.strip(".")
 
             if not cleaned_record_name.endswith(cleaned_hosted_zone_name):
-                error_msg = f"""
-                An error occurred (InvalidChangeBatch) when calling the ChangeResourceRecordSets operation:
-                RRSet with DNS name {record_set["Name"]} is not permitted in zone {the_zone.name}
-                """
-                return error_msg
+                raise DnsNameInvalidForZone(
+                    name=record_set["Name"], zone_name=the_zone.name
+                )
 
             if not record_set["Name"].endswith("."):
                 record_set["Name"] += "."
@@ -567,7 +574,7 @@ class Route53Backend(BaseBackend):
                     the_zone.delete_rrset_by_id(record_set["SetIdentifier"])
                 else:
                     the_zone.delete_rrset(record_set)
-        return None
+            the_zone.rr_changes.append(original_change)
 
     def list_hosted_zones(self):
         return self.zones.values()
@@ -612,7 +619,7 @@ class Route53Backend(BaseBackend):
 
         return zone_list
 
-    def get_hosted_zone(self, id_):
+    def get_hosted_zone(self, id_) -> FakeZone:
         the_zone = self.zones.get(id_.replace("/hostedzone/", ""))
         if not the_zone:
             raise NoSuchHostedZone(id_)

moto/route53/responses.py

@@ -219,9 +219,7 @@ class Route53(BaseResponse):
             if effective_rr_count > 1000:
                 raise InvalidChangeBatch
 
-            error_msg = self.backend.change_resource_record_sets(zoneid, change_list)
-            if error_msg:
-                return 400, headers, error_msg
+            self.backend.change_resource_record_sets(zoneid, change_list)
 
             return 200, headers, CHANGE_RRSET_RESPONSE
 

tests/test_route53/test_route53.py

@@ -1118,6 +1118,37 @@ def test_change_resource_record_invalid_action_value():
     len(response["ResourceRecordSets"]).should.equal(1)
 
 
+@mock_route53
+def test_change_resource_record_set_twice():
+    ZONE = "cname.local"
+    FQDN = f"test.{ZONE}"
+    FQDN_TARGET = "develop.domain.com"
+
+    client = boto3.client("route53", region_name="us-east-1")
+    zone_id = client.create_hosted_zone(
+        Name=ZONE, CallerReference="ref", DelegationSetId="string"
+    )["HostedZone"]["Id"]
+    changes = {
+        "Changes": [
+            {
+                "Action": "CREATE",
+                "ResourceRecordSet": {
+                    "Name": FQDN,
+                    "Type": "CNAME",
+                    "TTL": 600,
+                    "ResourceRecords": [{"Value": FQDN_TARGET}],
+                },
+            }
+        ]
+    }
+    client.change_resource_record_sets(HostedZoneId=zone_id, ChangeBatch=changes)
+
+    with pytest.raises(ClientError) as exc:
+        client.change_resource_record_sets(HostedZoneId=zone_id, ChangeBatch=changes)
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("InvalidChangeBatch")
+
+
 @mock_route53
 def test_list_resource_record_sets_name_type_filters():
     conn = boto3.client("route53", region_name="us-east-1")