Repository: getmoto/moto
Pull request #8367 (2024-12-05 16:31:26)

Validate RDS snapshot identifiers
...and improve the validation of RDS instance identifiers.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/8367?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.59%. Comparing base [(`549d4e2`)](https://app.codecov.io/gh/getmoto/moto/commit/549d4e2c5c565dadfdcf86d435aecf451c253cf9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`a3abad0`)](https://app.codecov.io/gh/getmoto/moto/commit/a3abad0d98974d3d2456753efc7a245ee766a2de?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

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


```diff
@@           Coverage Diff           @@
##           master    #8367   +/-   ##
=======================================
  Coverage   94.58%   94.59%           
=======================================
  Files        1159     1159           
  Lines      101297   101317   +20     
=======================================
+ Hits        95814    95836   +22     
+ Misses       5483     5481    -2     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/8367/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/8367/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `28.88% <86.36%> (+0.01%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/8367/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `94.56% <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/8367?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).

snordhausen: @bpandola I tried your idea of doing the checks directly in `responses.py` and I agree this looks much cleaner.

Patch:
diff --git a/moto/rds/exceptions.py b/moto/rds/exceptions.py
index 1829bc3f0..f76efba4b 100644
--- a/moto/rds/exceptions.py
+++ b/moto/rds/exceptions.py
@@ -212,6 +212,23 @@ class InvalidDBInstanceIdentifier(InvalidParameterValue):
         )
 
 
+class InvalidDBSnapshotIdentifier(InvalidParameterValue):
+    def __init__(self, snapshot_identifier: str, parameter_name: str) -> None:
+        if snapshot_identifier == "":
+            exception_text = f"The parameter {parameter_name} must be provided and must not be blank."
+        elif not snapshot_identifier[0].isalpha():
+            # On AWS, this error message seems to be triggered when the first character is invalid.
+            # The two spaces before the snapshot_identifier are what AWS produces!
+            exception_text = f"Invalid snapshot identifier:  {snapshot_identifier}"
+        else:
+            exception_text = (
+                f"The parameter {parameter_name} is not a valid identifier. "
+                "Identifiers must begin with a letter; must contain only ASCII letters, digits, and hyphens; "
+                "and must not end with a hyphen or contain two consecutive hyphens."
+            )
+        super().__init__(exception_text)
+
+
 class InvalidDBInstanceEngine(InvalidParameterCombination):
     def __init__(self, instance_engine: str, cluster_engine: str) -> None:
         super().__init__(
diff --git a/moto/rds/models.py b/moto/rds/models.py
index a65d69ad3..ddefed8a3 100644
--- a/moto/rds/models.py
+++ b/moto/rds/models.py
@@ -39,6 +39,7 @@ from .exceptions import (
     InvalidDBInstanceEngine,
     InvalidDBInstanceIdentifier,
     InvalidDBInstanceStateError,
+    InvalidDBSnapshotIdentifier,
     InvalidExportSourceStateError,
     InvalidGlobalClusterStateFault,
     InvalidParameterCombination,
@@ -2719,12 +2720,34 @@ class RDSBackend(BaseBackend):
         # # Must contain from 1 to 63 letters, numbers, or hyphens.
         # # First character must be a letter.
         # # Can't end with a hyphen or contain two consecutive hyphens.
-        if re.match(
-            "^(?!.*--)([a-zA-Z]?[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])$", db_identifier
+        if (
+            re.match(
+                "^(?!.*--)([a-zA-Z]?[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])$", db_identifier
+            )
+            and db_identifier[0].isalpha()
         ):
             return
         raise InvalidDBInstanceIdentifier
 
+    @staticmethod
+    def validate_db_snapshot_identifier(
+        db_snapshot_identifier: str, parameter_name: str
+    ) -> None:
+        # https://docs.aws.amazon.com/AmazonRDS/latest/APIReference/API_CreateDBSnapshot.html
+        # Constraints:
+        # # Must contain from 1 to 255 letters, numbers, or hyphens.
+        # # First character must be a letter.
+        # # Can't end with a hyphen or contain two consecutive hyphens.
+        if (
+            re.match(
+                "^(?!.*--)([a-zA-Z]?[a-zA-Z0-9-]{0,253}[a-zA-Z0-9])$",
+                db_snapshot_identifier,
+            )
+            and db_snapshot_identifier[0].isalpha()
+        ):
+            return
+        raise InvalidDBSnapshotIdentifier(db_snapshot_identifier, parameter_name)
+
     def describe_orderable_db_instance_options(
         self, engine: str, engine_version: str
     ) -> List[Dict[str, Any]]:
diff --git a/moto/rds/responses.py b/moto/rds/responses.py
index fabafe7fa..a972a2473 100644
--- a/moto/rds/responses.py
+++ b/moto/rds/responses.py
@@ -277,6 +277,11 @@ class RDSResponse(BaseResponse):
     def delete_db_instance(self) -> str:
         db_instance_identifier = self._get_param("DBInstanceIdentifier")
         db_snapshot_name = self._get_param("FinalDBSnapshotIdentifier")
+        if db_snapshot_name is not None:
+            self.backend.validate_db_snapshot_identifier(
+                db_snapshot_name, parameter_name="FinalDBSnapshotIdentifier"
+            )
+
         database = self.backend.delete_db_instance(
             db_instance_identifier, db_snapshot_name
         )
@@ -293,8 +298,13 @@ class RDSResponse(BaseResponse):
         db_instance_identifier = self._get_param("DBInstanceIdentifier")
         db_snapshot_identifier = self._get_param("DBSnapshotIdentifier")
         tags = self.unpack_list_params("Tags", "Tag")
+        self.backend.validate_db_snapshot_identifier(
+            db_snapshot_identifier, parameter_name="DBSnapshotIdentifier"
+        )
         snapshot = self.backend.create_db_snapshot(
-            db_instance_identifier, db_snapshot_identifier, tags=tags
+            db_instance_identifier,
+            db_snapshot_identifier,
+            tags=tags,
         )
         template = self.response_template(CREATE_SNAPSHOT_TEMPLATE)
         return template.render(snapshot=snapshot)
@@ -304,6 +314,10 @@ class RDSResponse(BaseResponse):
         target_snapshot_identifier = self._get_param("TargetDBSnapshotIdentifier")
         tags = self.unpack_list_params("Tags", "Tag")
         copy_tags = self._get_param("CopyTags")
+        self.backend.validate_db_snapshot_identifier(
+            target_snapshot_identifier, parameter_name="TargetDBSnapshotIdentifier"
+        )
+
         snapshot = self.backend.copy_db_snapshot(
             source_snapshot_identifier, target_snapshot_identifier, tags, copy_tags
         )
@@ -378,6 +392,11 @@ class RDSResponse(BaseResponse):
     def stop_db_instance(self) -> str:
         db_instance_identifier = self._get_param("DBInstanceIdentifier")
         db_snapshot_identifier = self._get_param("DBSnapshotIdentifier")
+        if db_snapshot_identifier is not None:
+            self.backend.validate_db_snapshot_identifier(
+                db_snapshot_identifier, parameter_name="DBSnapshotIdentifier"
+            )
+
         database = self.backend.stop_db_instance(
             db_instance_identifier, db_snapshot_identifier
         )


Test patch:
diff --git a/tests/test_rds/test_rds.py b/tests/test_rds/test_rds.py
index 7da065ace..4dad60da9 100644
--- a/tests/test_rds/test_rds.py
+++ b/tests/test_rds/test_rds.py
@@ -7,6 +7,8 @@ from botocore.exceptions import ClientError
 
 from moto import mock_aws
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
+from moto.rds.exceptions import InvalidDBInstanceIdentifier, InvalidDBSnapshotIdentifier
+from moto.rds.models import RDSBackend
 from tests import aws_verified
 
 DEFAULT_REGION = "us-west-2"
@@ -2778,6 +2780,46 @@ def test_create_db_instance_with_availability_zone():
     assert resp["DBInstances"][0]["AvailabilityZone"] == availability_zone
 
 
+invalid_identifiers = ("-foo", "foo-", "2foo", "foo--bar", "", "foo_bar")
+invalid_db_identifiers = invalid_identifiers + ("x" * 64,)
+invalid_db_snapshot_identifiers = invalid_identifiers + ("x" * 256,)
+
+
+@pytest.mark.parametrize("invalid_db_identifier", invalid_db_identifiers)
+def test_validate_db_identifier_backend_invalid(invalid_db_identifier):
+    with pytest.raises(InvalidDBInstanceIdentifier):
+        RDSBackend._validate_db_identifier(invalid_db_identifier)
+
+
+@pytest.mark.parametrize(
+    "invalid_db_snapshot_identifier", invalid_db_snapshot_identifiers
+)
+def test_validate_db_snapshot_identifier_backend_invalid(
+    invalid_db_snapshot_identifier,
+):
+    with pytest.raises(InvalidDBSnapshotIdentifier):
+        RDSBackend.validate_db_snapshot_identifier(
+            invalid_db_snapshot_identifier, "DBSnapshotIdentifier"
+        )
+
+
+valid_identifiers = ("f", "foo", "FOO", "FOO-bar-123")
+valid_db_identifiers = valid_identifiers + ("x" * 63,)
+valid_db_snapshot_identifiers = valid_identifiers + ("x" * 255,)
+
+
+@pytest.mark.parametrize("valid_db_identifier", valid_db_identifiers)
+def test_validate_db_identifier_backend_valid(valid_db_identifier):
+    RDSBackend._validate_db_identifier(valid_db_identifier)
+
+
+@pytest.mark.parametrize("valid_db_snapshot_identifier", valid_db_snapshot_identifiers)
+def test_validate_db_snapshot_identifier_backend_valid(valid_db_snapshot_identifier):
+    RDSBackend.validate_db_snapshot_identifier(
+        valid_db_snapshot_identifier, "DBSnapshotIdentifier"
+    )
+
+
 @mock_aws
 def test_validate_db_identifier():
     client = boto3.client("rds", region_name=DEFAULT_REGION)
@@ -2817,6 +2859,97 @@ def test_validate_db_identifier():
     validation_helper(exc)
 
 
+@mock_aws
+def test_validate_db_snapshot_identifier_different_operations():
+    client = boto3.client("rds", region_name=DEFAULT_REGION)
+    db_instance_identifier = "valid-identifier"
+    valid_db_snapshot_identifier = "valid"
+    invalid_db_snapshot_identifier = "--invalid--"
+
+    client.create_db_instance(
+        DBInstanceIdentifier=db_instance_identifier,
+        Engine="postgres",
+        DBName="staging-postgres",
+        DBInstanceClass="db.m1.small",
+    )
+
+    expected_message = f"Invalid snapshot identifier:  {invalid_db_snapshot_identifier}"
+    with pytest.raises(ClientError) as exc:
+        client.create_db_snapshot(
+            DBSnapshotIdentifier=invalid_db_snapshot_identifier,
+            DBInstanceIdentifier=db_instance_identifier,
+        )
+    snapshot_validation_helper(exc, expected_message)
+
+    client.create_db_snapshot(
+        DBSnapshotIdentifier=valid_db_snapshot_identifier,
+        DBInstanceIdentifier=db_instance_identifier,
+    )
+
+    with pytest.raises(ClientError) as exc:
+        client.copy_db_snapshot(
+            SourceDBSnapshotIdentifier=valid_db_snapshot_identifier,
+            TargetDBSnapshotIdentifier=invalid_db_snapshot_identifier,
+        )
+    snapshot_validation_helper(exc, expected_message)
+
+    with pytest.raises(ClientError) as exc:
+        client.stop_db_instance(
+            DBInstanceIdentifier=db_instance_identifier,
+            DBSnapshotIdentifier=invalid_db_snapshot_identifier,
+        )
+    snapshot_validation_helper(exc, expected_message)
+
+    with pytest.raises(ClientError) as exc:
+        client.delete_db_instance(
+            DBInstanceIdentifier=db_instance_identifier,
+            FinalDBSnapshotIdentifier=invalid_db_snapshot_identifier,
+        )
+    snapshot_validation_helper(exc, expected_message)
+
+
+# Depending on what is wrong with the snapshot_identifier, AWS produces different error messages.
+snapshot_identifier_data = (
+    ("", "The parameter DBSnapshotIdentifier must be provided and must not be blank."),
+    ("-invalid", "Invalid snapshot identifier:  -invalid"),
+    (
+        "in--valid",
+        (
+            "The parameter DBSnapshotIdentifier is not a valid identifier. "
+            "Identifiers must begin with a letter; must contain only ASCII "
+            "letters, digits, and hyphens; and must not end with a hyphen "
+            "or contain two consecutive hyphens."
+        ),
+    ),
+)
+
+
+@pytest.mark.parametrize(
+    "invalid_db_snapshot_identifier,expected_message",
+    snapshot_identifier_data,
+    ids=("empty", "invalid_first_character", "default_message"),
+)
+@mock_aws
+def test_validate_db_snapshot_identifier_different_error_messages(
+    invalid_db_snapshot_identifier, expected_message
+):
+    client = boto3.client("rds", region_name=DEFAULT_REGION)
+    db_instance_identifier = "valid-identifier"
+    client.create_db_instance(
+        DBInstanceIdentifier=db_instance_identifier,
+        Engine="postgres",
+        DBName="staging-postgres",
+        DBInstanceClass="db.m1.small",
+    )
+
+    with pytest.raises(ClientError) as exc:
+        client.create_db_snapshot(
+            DBSnapshotIdentifier=invalid_db_snapshot_identifier,
+            DBInstanceIdentifier=db_instance_identifier,
+        )
+    snapshot_validation_helper(exc, expected_message)
+
+
 @mock_aws
 def test_createdb_instance_engine_with_invalid_value():
     client = boto3.client("rds", region_name=DEFAULT_REGION)
@@ -2951,3 +3084,9 @@ def validation_helper(exc):
         "letters, digits, and hyphens; "
         "and must not end with a hyphen or contain two consecutive hyphens."
     )
+
+
+def snapshot_validation_helper(exc, expected_message):
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterValue"
+    assert err["Message"] == expected_message