feat[route53]: Validate action value in change_resource_record_sets. (#5222)

moto/route53/exceptions.py

@@ -15,11 +15,27 @@ class InvalidInput(Route53ClientError):
 
     code = 400
 
-    def __init__(self):
+    def __init__(self, message: str):
-        message = "The ARN for the CloudWatch Logs log group is invalid"
         super().__init__("InvalidInput", message)
 
 
+class InvalidCloudWatchArn(InvalidInput):
+    def __init__(
+        self,
+    ):
+        message = "The ARN for the CloudWatch Logs log group is invalid"
+        super().__init__(message)
+
+
+class InvalidActionValue(InvalidInput):
+    def __init__(self, value: str):
+        message = (
+            f"Invalid XML ; cvc-enumeration-valid: Value '{value}' is not facet-valid"
+            " with respect to enumeration '[CREATE, DELETE, UPSERT]'. It must be a value from the enumeration."
+        )
+        super().__init__(message)
+
+
 class InvalidPaginationToken(Route53ClientError):
     """Bad NextToken specified when listing query logging configs."""
 

moto/route53/models.py

@@ -10,7 +10,8 @@ from jinja2 import Template
 
 from moto.route53.exceptions import (
     HostedZoneNotEmpty,
-    InvalidInput,
+    InvalidActionValue,
+    InvalidCloudWatchArn,
     LastVPCAssociation,
     NoSuchCloudWatchLogsLogGroup,
     NoSuchDelegationSet,
@@ -501,6 +502,10 @@ class Route53Backend(BaseBackend):
         the_zone = self.get_hosted_zone(zoneid)
         for value in change_list:
             action = value["Action"]
+
+            if action not in ("CREATE", "UPSERT", "DELETE"):
+                raise InvalidActionValue(action)
+
             record_set = value["ResourceRecordSet"]
 
             cleaned_record_name = record_set["Name"].strip(".")
@@ -632,12 +637,12 @@ class Route53Backend(BaseBackend):
     def _validate_arn(region, arn):
         match = re.match(rf"arn:aws:logs:{region}:\d{{12}}:log-group:.+", arn)
         if not arn or not match:
-            raise InvalidInput()
+            raise InvalidCloudWatchArn()
 
         # The CloudWatch Logs log group must be in the "us-east-1" region.
         match = re.match(r"^(?:[^:]+:){3}(?P<region>[^:]+).*", arn)
         if match.group("region") != "us-east-1":
-            raise InvalidInput()
+            raise InvalidCloudWatchArn()
 
     def create_query_logging_config(self, region, hosted_zone_id, log_group_arn):
         """Process the create_query_logging_config request."""

tests/test_route53/test_route53.py

@@ -1268,6 +1268,51 @@ def test_change_resource_record_invalid():
     len(response["ResourceRecordSets"]).should.equal(1)
 
 
+@mock_route53
+def test_change_resource_record_invalid_action_value():
+    conn = boto3.client("route53", region_name="us-east-1")
+    conn.create_hosted_zone(
+        Name="db.",
+        CallerReference=str(hash("foo")),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="db"),
+    )
+
+    zones = conn.list_hosted_zones_by_name(DNSName="db.")
+    len(zones["HostedZones"]).should.equal(1)
+    zones["HostedZones"][0]["Name"].should.equal("db.")
+    hosted_zone_id = zones["HostedZones"][0]["Id"]
+
+    invalid_a_record_payload = {
+        "Comment": "this should fail",
+        "Changes": [
+            {
+                "Action": "INVALID_ACTION",
+                "ResourceRecordSet": {
+                    "Name": "prod.scooby.doo",
+                    "Type": "A",
+                    "TTL": 10,
+                    "ResourceRecords": [{"Value": "127.0.0.1"}],
+                },
+            }
+        ],
+    }
+
+    with pytest.raises(botocore.exceptions.ClientError) as exc:
+        conn.change_resource_record_sets(
+            HostedZoneId=hosted_zone_id, ChangeBatch=invalid_a_record_payload
+        )
+
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("InvalidInput")
+    err["Message"].should.equal(
+        "Invalid XML ; cvc-enumeration-valid: Value 'INVALID_ACTION' is not facet-valid"
+        " with respect to enumeration '[CREATE, DELETE, UPSERT]'. It must be a value from the enumeration."
+    )
+
+    response = conn.list_resource_record_sets(HostedZoneId=hosted_zone_id)
+    len(response["ResourceRecordSets"]).should.equal(1)
+
+
 @mock_route53
 def test_list_resource_record_sets_name_type_filters():
     conn = boto3.client("route53", region_name="us-east-1")