Repository: getmoto/moto
Issue category: Feature Request

Issue:
Feature: Determine if managed master password is enable on RDS if describe_db_clusters is used
I am currently working on a script which upgrades RDS instances with Blue-Green Deployments. Right now this requires to disable the Managed Master Password before creating the Blue-Green Deployment, so the script includes setting the Masterpassword statically and after completing the Upgrade, switching back to Managed Master Password.

Right now there is no way to tell if these switches work as expected, since `describe_db_clusters` does not return any information whether the master secret is managed or not. In the original boto3 response, `describe_db_clusters` contain the dict `MasterUserSecret` with further information about the Secret.

I happily contribute this feature if desired.


Hints:
Hi @lukaalba, a PR would be very welcome!

There is some generic info on how to contribute to Moto here:
https://docs.getmoto.org/en/latest/docs/contributing/index.html

The exact response of the `describe_db_clusters`-method is determined here, so the template used there would need some changes:
https://github.com/getmoto/moto/blob/bf49e22f41be7444dc20e05dfa5dd13f25672eba/moto/rds/responses.py#L252

(I haven't looked into it, but there are potentially also changes necessary to the `create_db_clusters`/`update_db_clusters`-methods to ensure that we store/manage that value in the first place.)

Please let me know if you have any questions!

Patch:
diff --git a/moto/rds/models.py b/moto/rds/models.py
index 51229c102ea9..e1ab97148af4 100644
--- a/moto/rds/models.py
+++ b/moto/rds/models.py
@@ -330,6 +330,14 @@ def __init__(
         else:
             self.master_user_password = kwargs.get("master_user_password")  # type: ignore
 
+        self.master_user_secret_kms_key_id = kwargs.get("master_user_secret_kms_key_id")
+        self.manage_master_user_password = kwargs.get(
+            "manage_master_user_password", False
+        )
+        self.master_user_secret_status = kwargs.get(
+            "master_user_secret_status", "active"
+        )
+
         self.availability_zones = kwargs.get("availability_zones")
         if not self.availability_zones:
             self.availability_zones = [
@@ -491,6 +499,15 @@ def enable_http_endpoint(self, val: Optional[bool]) -> None:
                 ]:
                     self._enable_http_endpoint = val
 
+    def master_user_secret(self) -> Dict[str, Any]:
+        return {
+            "secret_arn": f"arn:{self.partition}:secretsmanager:{self.region}:{self.account_id}:secret:rds!{self.name}",
+            "secret_status": self.master_user_secret_status,
+            "kms_key_id": self.master_user_secret_kms_key_id
+            if self.master_user_secret_kms_key_id is not None
+            else f"arn:{self.partition}:kms:{self.region}:{self.account_id}:key/{self.name}",
+        }
+
     def get_cfg(self) -> Dict[str, Any]:
         cfg = self.__dict__
         cfg.pop("backend")
@@ -603,6 +620,14 @@ def to_xml(self, initial: bool = False) -> str:
               <GlobalClusterIdentifier>{{ cluster.global_cluster_identifier }}</GlobalClusterIdentifier>
               {%- endif -%}
               {%- if cluster.replication_source_identifier -%}<ReplicationSourceIdentifier>{{ cluster.replication_source_identifier }}</ReplicationSourceIdentifier>{%- endif -%}
+              {%- if cluster.manage_master_user_password -%}
+              <MasterUserSecret>
+                {% set master_user_secret = cluster.master_user_secret() %}
+                {% if "secret_arn" in master_user_secret %}<SecretArn>{{ master_user_secret["secret_arn"] }}</SecretArn>{% endif %}
+                {% if "secret_status" in master_user_secret %}<SecretStatus>{{ master_user_secret["secret_status"] }}</SecretStatus>{% endif %}
+                {% if "kms_key_id" in master_user_secret %}<KmsKeyId>{{ master_user_secret["kms_key_id"] }}</KmsKeyId>{% endif %}
+              </MasterUserSecret>
+              {%- endif -%}
             </DBCluster>"""
         )
         return template.render(cluster=self, status=status)
@@ -776,6 +801,13 @@ def __init__(
             self.storage_type = DBInstance.default_storage_type(iops=self.iops)
         self.master_username = kwargs.get("master_username")
         self.master_user_password = kwargs.get("master_user_password")
+        self.master_user_secret_kms_key_id = kwargs.get("master_user_secret_kms_key_id")
+        self.master_user_secret_status = kwargs.get(
+            "master_user_secret_status", "active"
+        )
+        self.manage_master_user_password = kwargs.get(
+            "manage_master_user_password", False
+        )
         self.auto_minor_version_upgrade = kwargs.get("auto_minor_version_upgrade")
         if self.auto_minor_version_upgrade is None:
             self.auto_minor_version_upgrade = True
@@ -919,6 +951,15 @@ def default_db_parameter_group_details(self) -> Tuple[str, str]:
 
         return db_family, f"default.{db_family}"
 
+    def master_user_secret(self) -> Dict[str, Any]:
+        return {
+            "secret_arn": f"arn:{self.partition}:secretsmanager:{self.region}:{self.account_id}:secret:rds!{self.name}",
+            "secret_status": self.master_user_secret_status,
+            "kms_key_id": self.master_user_secret_kms_key_id
+            if self.master_user_secret_kms_key_id is not None
+            else f"arn:{self.partition}:kms:{self.region}:{self.account_id}:key/{self.name}",
+        }
+
     def to_xml(self) -> str:
         template = Template(
             """<DBInstance>
@@ -1041,6 +1082,14 @@ def to_xml(self) -> str:
               {%- endfor -%}
               </TagList>
               <DeletionProtection>{{ 'true' if database.deletion_protection else 'false' }}</DeletionProtection>
+              {%- if database.manage_master_user_password -%}
+              <MasterUserSecret>
+                {% set master_user_secret = database.master_user_secret() %}
+                {% if "secret_arn" in master_user_secret %}<SecretArn>{{ master_user_secret["secret_arn"] }}</SecretArn>{% endif %}
+                {% if "secret_status" in master_user_secret %}<SecretStatus>{{ master_user_secret["secret_status"] }}</SecretStatus>{% endif %}
+                {% if "kms_key_id" in master_user_secret %}<KmsKeyId>{{ master_user_secret["kms_key_id"] }}</KmsKeyId>{% endif %}
+              </MasterUserSecret>
+              {%- endif -%}
             </DBInstance>"""
         )
         return template.render(database=self)
@@ -1935,8 +1984,23 @@ def modify_db_instance(
         )
         if msg:
             raise RDSClientError("InvalidParameterValue", msg)
+        if db_kwargs.get("rotate_master_user_password") and db_kwargs.get(
+            "apply_immediately"
+        ):
+            db_kwargs["master_user_secret_status"] = "rotating"
+        if db_kwargs.get("rotate_master_user_password") and not db_kwargs.get(
+            "apply_immediately"
+        ):
+            raise RDSClientError(
+                "InvalidParameterCombination",
+                "You must specify apply immediately when rotating the master user password.",
+            )
         database.update(db_kwargs)
-        return database
+        initial_state = copy.deepcopy(database)
+        database.master_user_secret_status = (
+            "active"  # already set the final state in the background
+        )
+        return initial_state
 
     def reboot_db_instance(self, db_instance_identifier: str) -> DBInstance:
         return self.describe_db_instances(db_instance_identifier)[0]
@@ -2385,6 +2449,18 @@ def modify_db_cluster(self, kwargs: Dict[str, Any]) -> DBCluster:
         cluster = self.clusters[cluster_id]
         del self.clusters[cluster_id]
 
+        if kwargs.get("rotate_master_user_password") and kwargs.get(
+            "apply_immediately"
+        ):
+            kwargs["master_user_secret_status"] = "rotating"
+        elif kwargs.get("rotate_master_user_password") and not kwargs.get(
+            "apply_immediately"
+        ):
+            raise RDSClientError(
+                "InvalidParameterCombination",
+                "You must specify apply immediately when rotating the master user password.",
+            )
+
         kwargs["db_cluster_identifier"] = kwargs.pop("new_db_cluster_identifier")
         for k, v in kwargs.items():
             if k == "db_cluster_parameter_group_name":
@@ -2403,7 +2479,11 @@ def modify_db_cluster(self, kwargs: Dict[str, Any]) -> DBCluster:
         self.clusters[cluster_id] = cluster
 
         initial_state = copy.deepcopy(cluster)  # Return status=creating
-        cluster.status = "available"  # Already set the final status in the background
+
+        # Already set the final status in the background
+        cluster.status = "available"
+        cluster.master_user_secret_status = "active"
+
         return initial_state
 
     def promote_read_replica_db_cluster(self, db_cluster_identifier: str) -> DBCluster:
diff --git a/moto/rds/responses.py b/moto/rds/responses.py
index 7b8911369c9c..42acfdd96364 100644
--- a/moto/rds/responses.py
+++ b/moto/rds/responses.py
@@ -48,6 +48,13 @@ def _get_db_kwargs(self) -> Dict[str, Any]:
             "kms_key_id": self._get_param("KmsKeyId"),
             "master_user_password": self._get_param("MasterUserPassword"),
             "master_username": self._get_param("MasterUsername"),
+            "manage_master_user_password": self._get_bool_param(
+                "ManageMasterUserPassword"
+            ),
+            "master_user_secret_kms_key_id": self._get_param(
+                "MasterUserSecretKmsKeyId"
+            ),
+            "rotate_master_user_password": self._get_param("RotateMasterUserPassword"),
             "multi_az": self._get_bool_param("MultiAZ"),
             "option_group_name": self._get_param("OptionGroupName"),
             "port": self._get_param("Port"),
@@ -68,6 +75,7 @@ def _get_db_kwargs(self) -> Dict[str, Any]:
             ),
             "tags": list(),
             "deletion_protection": self._get_bool_param("DeletionProtection"),
+            "apply_immediately": self._get_bool_param("ApplyImmediately"),
         }
         args["tags"] = self.unpack_list_params("Tags", "Tag")
         return args
@@ -105,6 +113,13 @@ def _get_modify_db_cluster_kwargs(self) -> Dict[str, Any]:
             "kms_key_id": self._get_param("KmsKeyId"),
             "master_user_password": self._get_param("MasterUserPassword"),
             "master_username": self._get_param("MasterUsername"),
+            "manage_master_user_password": self._get_bool_param(
+                "ManageMasterUserPassword"
+            ),
+            "master_user_secret_kms_key_id": self._get_param(
+                "MasterUserSecretKmsKeyId"
+            ),
+            "rotate_master_user_password": self._get_param("RotateMasterUserPassword"),
             "multi_az": self._get_bool_param("MultiAZ"),
             "option_group_name": self._get_param("OptionGroupName"),
             "port": self._get_param("Port"),
@@ -123,6 +138,7 @@ def _get_modify_db_cluster_kwargs(self) -> Dict[str, Any]:
             ),
             "tags": list(),
             "deletion_protection": self._get_bool_param("DeletionProtection"),
+            "apply_immediately": self._get_bool_param("ApplyImmediately"),
         }
         args["tags"] = self.unpack_list_params("Tags", "Tag")
         return args
@@ -186,6 +202,12 @@ def _get_db_cluster_kwargs(self) -> Dict[str, Any]:
             "storage_type": self._get_param("StorageType"),
             "kms_key_id": self._get_param("KmsKeyId"),
             "master_username": self._get_param("MasterUsername"),
+            "manage_master_user_password": self._get_bool_param(
+                "ManageMasterUserPassword"
+            ),
+            "master_user_secret_kms_key_id": self._get_param(
+                "MasterUserSecretKmsKeyId"
+            ),
             "master_user_password": self._get_param("MasterUserPassword"),
             "network_type": self._get_param("NetworkType"),
             "port": self._get_param("Port"),


Test patch:
diff --git a/tests/test_rds/test_rds.py b/tests/test_rds/test_rds.py
index 68913255f9e5..e9de65a134fb 100644
--- a/tests/test_rds/test_rds.py
+++ b/tests/test_rds/test_rds.py
@@ -580,6 +580,104 @@ def test_modify_db_instance():
     assert inst["EnabledCloudwatchLogsExports"] == ["error"]
 
 
+@pytest.mark.parametrize("with_custom_kms_key", [True, False])
+@mock_aws
+def test_modify_db_instance_manage_master_user_password(with_custom_kms_key: bool):
+    conn = boto3.client("rds", region_name=DEFAULT_REGION)
+    db_id = "db-id"
+
+    custom_kms_key = f"arn:aws:kms:{DEFAULT_REGION}:123456789012:key/abcd1234-56ef-78gh-90ij-klmnopqrstuv"
+    custom_kms_key_args = (
+        {"MasterUserSecretKmsKeyId": custom_kms_key} if with_custom_kms_key else {}
+    )
+
+    create_response = conn.create_db_instance(
+        DBInstanceIdentifier=db_id,
+        DBInstanceClass="postgres",
+        Engine="postgres",
+        MasterUsername="root",
+        MasterUserPassword="hunter21",
+        ManageMasterUserPassword=False,
+    )
+
+    modify_response = conn.modify_db_instance(
+        DBInstanceIdentifier=db_id, ManageMasterUserPassword=True, **custom_kms_key_args
+    )
+
+    describe_response = conn.describe_db_instances(
+        DBInstanceIdentifier=db_id,
+    )
+
+    revert_modification_response = conn.modify_db_instance(
+        DBInstanceIdentifier=db_id, ManageMasterUserPassword=False
+    )
+
+    assert create_response["DBInstance"].get("MasterUserSecret") is None
+    master_user_secret = modify_response["DBInstance"]["MasterUserSecret"]
+    assert len(master_user_secret.keys()) == 3
+    assert (
+        master_user_secret["SecretArn"]
+        == "arn:aws:secretsmanager:us-west-2:123456789012:secret:rds!db-id"
+    )
+    assert master_user_secret["SecretStatus"] == "active"
+    if with_custom_kms_key:
+        assert master_user_secret["KmsKeyId"] == custom_kms_key
+    else:
+        assert (
+            master_user_secret["KmsKeyId"]
+            == "arn:aws:kms:us-west-2:123456789012:key/db-id"
+        )
+    assert len(describe_response["DBInstances"][0]["MasterUserSecret"].keys()) == 3
+    assert (
+        modify_response["DBInstance"]["MasterUserSecret"]
+        == describe_response["DBInstances"][0]["MasterUserSecret"]
+    )
+    assert revert_modification_response["DBInstance"].get("MasterUserSecret") is None
+
+
+@pytest.mark.parametrize("with_apply_immediately", [True, False])
+@mock_aws
+def test_modify_db_instance_rotate_master_user_password(with_apply_immediately):
+    conn = boto3.client("rds", region_name=DEFAULT_REGION)
+    db_id = "db-id"
+
+    conn.create_db_instance(
+        DBInstanceIdentifier=db_id,
+        DBInstanceClass="postgres",
+        Engine="postgres",
+        MasterUsername="root",
+        MasterUserPassword="hunter21",
+        ManageMasterUserPassword=True,
+    )
+
+    if with_apply_immediately:
+        modify_response = conn.modify_db_instance(
+            DBInstanceIdentifier=db_id,
+            RotateMasterUserPassword=True,
+            ApplyImmediately=True,
+        )
+
+        describe_response = conn.describe_db_instances(
+            DBInstanceIdentifier=db_id,
+        )
+
+        assert (
+            modify_response["DBInstance"]["MasterUserSecret"]["SecretStatus"]
+            == "rotating"
+        )
+        assert (
+            describe_response["DBInstances"][0]["MasterUserSecret"]["SecretStatus"]
+            == "active"
+        )
+
+    else:
+        with pytest.raises(ClientError):
+            conn.modify_db_instance(
+                DBInstanceIdentifier=db_id,
+                RotateMasterUserPassword=True,
+            )
+
+
 @mock_aws
 def test_modify_db_instance_not_existent_db_parameter_group_name():
     conn = boto3.client("rds", region_name=DEFAULT_REGION)
diff --git a/tests/test_rds/test_rds_clusters.py b/tests/test_rds/test_rds_clusters.py
index f4fabf822450..a12fb38328f0 100644
--- a/tests/test_rds/test_rds_clusters.py
+++ b/tests/test_rds/test_rds_clusters.py
@@ -150,6 +150,103 @@ def test_modify_db_cluster_new_cluster_identifier():
     assert old_id not in clusters
 
 
+@pytest.mark.parametrize("with_custom_kms_key", [True, False])
+@mock_aws
+def test_modify_db_cluster_manage_master_user_password(with_custom_kms_key):
+    client = boto3.client("rds", region_name=RDS_REGION)
+    cluster_id = "cluster-id"
+    custom_kms_key = f"arn:aws:kms:{RDS_REGION}:123456789012:key/abcd1234-56ef-78gh-90ij-klmnopqrstuv"
+    custom_kms_key_args = (
+        {"MasterUserSecretKmsKeyId": custom_kms_key} if with_custom_kms_key else {}
+    )
+
+    create_response = client.create_db_cluster(
+        DBClusterIdentifier=cluster_id,
+        Engine="aurora-postgresql",
+        MasterUsername="root",
+        MasterUserPassword="hunter21",
+        ManageMasterUserPassword=False,
+    )
+
+    modify_response = client.modify_db_cluster(
+        DBClusterIdentifier=cluster_id,
+        ManageMasterUserPassword=True,
+        **custom_kms_key_args,
+    )
+
+    describe_response = client.describe_db_clusters(
+        DBClusterIdentifier=cluster_id,
+    )
+
+    revert_modification_response = client.modify_db_cluster(
+        DBClusterIdentifier=cluster_id, ManageMasterUserPassword=False
+    )
+
+    assert create_response["DBCluster"].get("MasterUserSecret") is None
+    master_user_secret = modify_response["DBCluster"]["MasterUserSecret"]
+    assert len(master_user_secret.keys()) == 3
+    assert (
+        master_user_secret["SecretArn"]
+        == "arn:aws:secretsmanager:eu-north-1:123456789012:secret:rds!cluster-id"
+    )
+    assert master_user_secret["SecretStatus"] == "active"
+    if with_custom_kms_key:
+        assert master_user_secret["KmsKeyId"] == custom_kms_key
+    else:
+        assert (
+            master_user_secret["KmsKeyId"]
+            == "arn:aws:kms:eu-north-1:123456789012:key/cluster-id"
+        )
+    assert len(describe_response["DBClusters"][0]["MasterUserSecret"].keys()) == 3
+    assert (
+        modify_response["DBCluster"]["MasterUserSecret"]
+        == describe_response["DBClusters"][0]["MasterUserSecret"]
+    )
+    assert revert_modification_response["DBCluster"].get("MasterUserSecret") is None
+
+
+@pytest.mark.parametrize("with_apply_immediately", [True, False])
+@mock_aws
+def test_modify_db_cluster_rotate_master_user_password(with_apply_immediately):
+    client = boto3.client("rds", region_name=RDS_REGION)
+    cluster_id = "cluster-id"
+
+    client.create_db_cluster(
+        DBClusterIdentifier=cluster_id,
+        Engine="aurora-postgresql",
+        MasterUsername="root",
+        MasterUserPassword="hunter21",
+        ManageMasterUserPassword=True,
+    )
+
+    if with_apply_immediately:
+        modify_response = client.modify_db_cluster(
+            DBClusterIdentifier=cluster_id,
+            RotateMasterUserPassword=True,
+            ApplyImmediately=True,
+        )
+
+        describe_response = client.describe_db_clusters(
+            DBClusterIdentifier=cluster_id,
+        )
+
+        assert (
+            modify_response["DBCluster"]["MasterUserSecret"]["SecretStatus"]
+            == "rotating"
+        )
+        assert (
+            describe_response["DBClusters"][0]["MasterUserSecret"]["SecretStatus"]
+            == "active"
+        )
+
+    else:
+        with pytest.raises(ClientError):
+            client.modify_db_cluster(
+                DBClusterIdentifier=cluster_id,
+                RotateMasterUserPassword=True,
+            )
+
+
 @mock_aws
 def test_create_db_cluster__verify_default_properties():
     client = boto3.client("rds", region_name=RDS_REGION)