Repository: getmoto/moto (python)

Issue:
[RDS] Unsupported `ManageMasterUserPassword` on create_db_cluster
According to [boto3 docs](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/rds/client/create_db_cluster.html), the `MasterUserPassword` can’t be specified if `ManageMasterUserPassword` is turned on.

As of v5.1.1, the `MasterUserPassword` is a mandatory field on the `create_db_cluster` function

Example test:
```
import boto3
from moto import mock_aws

@mock_aws
def test_create_db_cluster():
    rds = boto3.client('rds', region_name='us-east-1')

    # Create the DB cluster with a dummy password
    response = rds.create_db_cluster(
        DBClusterIdentifier='my-cluster',
        Engine='aurora-mysql',
        MasterUsername='myuser',
        ManageMasterUserPassword=True,
        DBClusterParameterGroupName='default-aurora-mysql57',
        VpcSecurityGroupIds=['sg-12345678'],
    )

test_create_db_cluster()
```

Result:
`botocore.exceptions.ClientError: An error occurred (InvalidParameterValue) when calling the CreateDBCluster operation: The parameter MasterUserPassword must be provided and must not be blank.`

Pull request description:
RDS: integrate with SecretsManager for managing master user password 
This PR builds on #8373, integrating the RDS and SecretsManager backends to provide an AWS managed secret for storing DBCluster/DBInstance master user passwords.

An internal `create_managed_secret` method has been added to the SecretsManager backend, allowing other services to easily create managed secrets.  The EventBridge backend, which had been doing its own simulation of a managed secret, has been updated to use this new method.

Comprehensive `aws_verified` tests covering the entire Managed Master User Password lifecycle are also included. 

Closes #8724



Patch:
diff --git a/moto/events/models.py b/moto/events/models.py
index dc54f52e8..f0281837f 100644
--- a/moto/events/models.py
+++ b/moto/events/models.py
@@ -823,18 +823,13 @@ class Connection(BaseModel):
                 secret_value.update(value)
         secret_value = CamelToUnderscoresWalker.parse(secret_value)
 
-        secret = secretsmanager_backend.create_secret(
-            name=f"events!connection/{connection_id}/auth",
+        secret = secretsmanager_backend.create_managed_secret(
+            service_name="events",
+            secret_id=f"events!connection/{connection_id}/auth",
             secret_string=json.dumps(secret_value),
-            replica_regions=[],
-            force_overwrite=False,
-            secret_binary=None,
             description=f"Auth parameters for Eventbridge connection {connection_id}",
-            tags=[{"Key": "aws:secretsmanager:owningService", "Value": "events"}],
-            kms_key_id=None,
-            client_request_token=None,
         )
-        self.secret_arn = json.loads(secret).get("ARN")
+        self.secret_arn = secret.arn
         self.arn = f"arn:{get_partition(region_name)}:events:{region_name}:{account_id}:connection/{connection_id}"
 
     def describe_short(self) -> Dict[str, Any]:
diff --git a/moto/rds/models.py b/moto/rds/models.py
index e034dca87..a70406596 100644
--- a/moto/rds/models.py
+++ b/moto/rds/models.py
@@ -5,6 +5,7 @@ import math
 import os
 import re
 import string
+import uuid
 from collections import OrderedDict, defaultdict
 from datetime import datetime
 from functools import lru_cache, partialmethod
@@ -28,6 +29,7 @@ from moto.core.utils import unix_time, utcnow
 from moto.ec2.models import ec2_backends
 from moto.kms.models import KmsBackend, kms_backends
 from moto.moto_api._internal import mock_random as random
+from moto.secretsmanager.models import FakeSecret, SecretsManagerBackend
 from moto.utilities.utils import ARN_PARTITION_REGEX, load_resource
 
 from .exceptions import (
@@ -374,6 +376,74 @@ class GlobalCluster(RDSBaseModel):
         return members
 
 
+class MasterUserSecret:
+    """Class to manage the master user password for RDS clusters/instances."""
+
+    def __init__(self, resource: DBCluster | DBInstance, kms_key_id: str | None):
+        self.resource = resource
+        self.kms = resource.backend.kms
+        self.secretsmanager = resource.backend.secretsmanager
+        self.secret = self._create_secret(kms_key_id)
+        self._status = "creating"
+
+    def delete_secret(self) -> None:
+        self.secretsmanager.delete_secret(
+            self.secret.arn,
+            recovery_window_in_days=None,
+            force_delete_without_recovery=True,
+        )
+
+    def rotate_secret(self) -> None:
+        self._status = "rotating"
+        self.secretsmanager.rotate_secret(self.secret.arn, rotate_immediately=True)
+
+    @property
+    def kms_key_id(self) -> str:
+        assert self.secret.kms_key_id is not None
+        key = self.kms.describe_key(self.secret.kms_key_id)
+        return key.arn
+
+    @property
+    def secret_arn(self) -> str:
+        return self.secret.arn
+
+    @property
+    def secret_status(self) -> str:
+        status_to_return = self._status
+        # Poor man's state manager - look into using moto's state manager.
+        if status_to_return in ["creating", "rotating"]:
+            self._status = "active"
+        return status_to_return
+
+    def _create_secret(self, kms_key_id: Optional[str]) -> FakeSecret:
+        secret = self.secretsmanager.create_managed_secret(
+            service_name="rds",
+            secret_id=self._generate_secret_name(),
+            secret_string="P@55w0rd!",
+            description=self._generate_secret_description(),
+            kms_key_id=kms_key_id,
+            tags=self._generate_secret_tags(),
+        )
+        return secret
+
+    def _generate_secret_name(self) -> str:
+        unique_id = str(uuid.uuid4())
+        secret_name = f"rds!{self.resource.resource_type}-{unique_id}"
+        return secret_name
+
+    def _generate_secret_description(self) -> str:
+        resource_type = self.resource.__class__.__name__[2:].lower()
+        description = f"The secret associated with the primary RDS DB {resource_type}: {self.resource.arn}"
+        return description
+
+    def _generate_secret_tags(self) -> list[dict[str, str]]:
+        resource_type = self.resource.__class__.__name__
+        tags = [
+            {"Key": f"aws:rds:primary{resource_type}Arn", "Value": self.resource.arn},
+        ]
+        return tags
+
+
 class DBCluster(RDSBaseModel):
     SUPPORTED_FILTERS = {
         "db-cluster-id": FilterDef(
@@ -409,6 +479,8 @@ class DBCluster(RDSBaseModel):
         vpc_security_group_ids: Optional[List[str]] = None,
         deletion_protection: Optional[bool] = False,
         kms_key_id: Optional[str] = None,
+        manage_master_user_password: Optional[bool] = False,
+        master_user_secret_kms_key_id: Optional[str] = None,
         **kwargs: Any,
     ):
         super().__init__(backend)
@@ -465,17 +537,13 @@ class DBCluster(RDSBaseModel):
             raise InvalidParameterValue(
                 "The parameter MasterUsername must be provided and must not be blank."
             )
-        else:
+        self.manage_master_user_password = manage_master_user_password
+        if self.manage_master_user_password:
+            self.master_user_secret = MasterUserSecret(
+                self, master_user_secret_kms_key_id
+            )
+        elif not self.global_cluster_identifier and not self.engine == "neptune":
             self.master_user_password = master_user_password or ""
-
-        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 = [
@@ -647,17 +715,6 @@ class DBCluster(RDSBaseModel):
     def http_endpoint_enabled(self) -> bool:
         return True if self.enable_http_endpoint else False
 
-    @property
-    def master_user_secret(self) -> Optional[Dict[str, Any]]:  # type: ignore[misc]
-        secret_info = {
-            "SecretArn": f"arn:{self.partition}:secretsmanager:{self.region}:{self.account_id}:secret:rds!{self.resource_id}",
-            "SecretStatus": self.master_user_secret_status,
-            "KmsKeyId": 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.resource_id}",
-        }
-        return secret_info if self.manage_master_user_password else None
-
     @property
     def db_cluster_parameter_group(self) -> str:
         return self.parameter_group
@@ -977,6 +1034,8 @@ class DBInstance(EventMixin, CloudFormationModel, RDSBaseModel):
         enable_cloudwatch_logs_exports: Optional[List[str]] = None,
         ca_certificate_identifier: str = "rds-ca-default",
         availability_zone: Optional[str] = None,
+        manage_master_user_password: Optional[bool] = False,
+        master_user_secret_kms_key_id: Optional[str] = None,
         **kwargs: Any,
     ) -> None:
         super().__init__(backend)
@@ -990,13 +1049,6 @@ class DBInstance(EventMixin, CloudFormationModel, RDSBaseModel):
             )
         self.log_file_manager = LogFileManager(self.engine)
         self.iops = iops
-        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 = auto_minor_version_upgrade
         self.db_instance_identifier = db_instance_identifier
         self.source_db_identifier = source_db_instance_identifier
@@ -1059,7 +1111,13 @@ class DBInstance(EventMixin, CloudFormationModel, RDSBaseModel):
             )
             self.license_model = license_model
             self.master_username = master_username
-            self.master_user_password = master_user_password
+            self.manage_master_user_password = manage_master_user_password
+            if self.manage_master_user_password:
+                self.master_user_secret = MasterUserSecret(
+                    self, master_user_secret_kms_key_id
+                )
+            else:
+                self.master_user_password = master_user_password
             self.preferred_backup_window = preferred_backup_window
             msg = valid_preferred_maintenance_window(
                 self.preferred_maintenance_window,
@@ -1261,17 +1319,6 @@ class DBInstance(EventMixin, CloudFormationModel, RDSBaseModel):
 
         return db_family, f"default.{db_family}"
 
-    @property
-    def master_user_secret(self) -> Dict[str, Any] | None:  # type: ignore[misc]
-        secret_info = {
-            "SecretArn": f"arn:{self.partition}:secretsmanager:{self.region}:{self.account_id}:secret:rds!{self.resource_id}",
-            "SecretStatus": self.master_user_secret_status,
-            "KmsKeyId": 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.resource_id}",
-        }
-        return secret_info if self.manage_master_user_password else None
-
     @property
     def address(self) -> str:
         return (
@@ -1354,7 +1401,24 @@ class DBInstance(EventMixin, CloudFormationModel, RDSBaseModel):
         self.is_replica = True
         self.replicas = []
 
-    def update(self, db_kwargs: Dict[str, Any]) -> None:
+    def update(
+        self,
+        manage_master_user_password: Optional[bool] = None,
+        master_user_secret_kms_key_id: Optional[str] = None,
+        rotate_master_user_password: Optional[bool] = None,
+        **db_kwargs: Dict[str, Any],
+    ) -> None:
+        if manage_master_user_password is True:
+            self.master_user_secret = MasterUserSecret(
+                self, master_user_secret_kms_key_id
+            )
+        elif manage_master_user_password is False:
+            self.master_user_secret.delete_secret()
+            del self.master_user_secret
+
+        if rotate_master_user_password is True:
+            self.master_user_secret.rotate_secret()
+
         for key, value in db_kwargs.items():
             if value is not None:
                 setattr(self, key, value)
@@ -1517,6 +1581,7 @@ class DBInstanceClustered(DBInstance):
         self.db_cluster_identifier: str = db_cluster_identifier
         self.is_cluster_writer = True if not self.cluster.members else False
         self.promotion_tier = promotion_tier
+        self.manage_master_user_password = False
 
     @property
     def allocated_storage(self) -> int:
@@ -2090,6 +2155,10 @@ class RDSBackend(BaseBackend):
     def kms(self) -> KmsBackend:
         return kms_backends[self.account_id][self.region_name]
 
+    @property
+    def secretsmanager(self) -> SecretsManagerBackend:
+        return self.get_backend("secretsmanager", self.region_name, self.account_id)  # type: ignore[return-value]
+
     def _validate_kms_key(self, kms_key_id: str) -> str:
         key = kms_key_id
         kms_backend = self.kms
@@ -2110,10 +2179,10 @@ class RDSBackend(BaseBackend):
 
     def get_backend(
         self,
-        service: Literal["kms"] | Literal["rds"],
+        service: Literal["kms"] | Literal["rds"] | Literal["secretsmanager"],
         region: str,
         account_id: Optional[str] = None,
-    ) -> KmsBackend | RDSBackend:
+    ) -> KmsBackend | RDSBackend | SecretsManagerBackend:
         from moto.backends import get_backend as get_moto_backend
 
         if account_id is None:
@@ -2304,7 +2373,7 @@ class RDSBackend(BaseBackend):
         database = self.databases[database_id]
         if database.is_replica:
             database.is_replica = False
-            database.update(db_kwargs)
+            database.update(**db_kwargs)
 
         return database
 
@@ -2317,7 +2386,7 @@ class RDSBackend(BaseBackend):
 
         # Shouldn't really copy here as the instance is duplicated. RDS replicas have different instances.
         replica = copy.copy(primary)
-        replica.update(db_kwargs)
+        replica.update(**db_kwargs)
         replica.set_as_replica()
         self.databases[database_id] = replica
         primary.add_replica(replica)
@@ -2399,10 +2468,7 @@ class RDSBackend(BaseBackend):
             )
             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"
         ):
@@ -2410,12 +2476,8 @@ class RDSBackend(BaseBackend):
                 "InvalidParameterCombination",
                 "You must specify apply immediately when rotating the master user password.",
             )
-        database.update(db_kwargs)
-        initial_state = copy.copy(database)
-        database.master_user_secret_status = (
-            "active"  # already set the final state in the background
-        )
-        return initial_state
+        database.update(**db_kwargs)
+        return database
 
     def reboot_db_instance(self, db_instance_identifier: str) -> DBInstance:
         return self.describe_db_instances(db_instance_identifier)[0]
@@ -2608,6 +2670,8 @@ class RDSBackend(BaseBackend):
                     snapshot_type="manual",
                 )
             database = self.databases.pop(db_instance_identifier)
+            if database.manage_master_user_password:
+                database.master_user_secret.delete_secret()
             if database.is_replica:
                 primary = self.find_db_from_id(database.source_db_instance_identifier)  # type: ignore
                 primary.remove_replica(database)
@@ -2998,7 +3062,7 @@ class RDSBackend(BaseBackend):
         if kwargs.get("rotate_master_user_password") and kwargs.get(
             "apply_immediately"
         ):
-            kwargs["master_user_secret_status"] = "rotating"
+            cluster.master_user_secret.rotate_secret()
         elif kwargs.get("rotate_master_user_password") and not kwargs.get(
             "apply_immediately"
         ):
@@ -3007,6 +3071,15 @@ class RDSBackend(BaseBackend):
                 "You must specify apply immediately when rotating the master user password.",
             )
 
+        if "manage_master_user_password" in kwargs:
+            manage_master_user_password = kwargs.pop("manage_master_user_password")
+            if manage_master_user_password:
+                kms_key_id = kwargs.pop("master_user_secret_kms_key_id", None)
+                cluster.master_user_secret = MasterUserSecret(cluster, kms_key_id)
+            else:
+                cluster.master_user_secret.delete_secret()
+                del cluster.master_user_secret
+
         kwargs["db_cluster_identifier"] = kwargs.pop("new_db_cluster_identifier", None)
         for k, v in kwargs.items():
             if k == "db_cluster_parameter_group_name":
@@ -3028,7 +3101,6 @@ class RDSBackend(BaseBackend):
 
         # Already set the final status in the background
         cluster.status = "available"
-        cluster.master_user_secret_status = "active"
 
         return initial_state
 
@@ -3174,7 +3246,8 @@ class RDSBackend(BaseBackend):
             global_id = cluster.global_cluster_identifier or ""
             if global_id in self.global_clusters:
                 self.remove_from_global_cluster(global_id, cluster_identifier)
-
+            if cluster.manage_master_user_password:
+                cluster.master_user_secret.delete_secret()
             if snapshot_name:
                 self.create_db_cluster_snapshot(cluster_identifier, snapshot_name)
             return self.clusters.pop(cluster_identifier)
@@ -3879,6 +3952,10 @@ class Event(XFormedAttributeAccessMixin):
             "Categories": ["creation"],
             "Message": "DB instance created",
         },
+        "DB_INSTANCE_RESET_MASTER_CREDENTIALS": {
+            "Categories": ["configuration change"],
+            "Message": "Reset master credentials",
+        },
         "DB_SNAPSHOT_CREATE_AUTOMATED_START": {
             "Categories": ["creation"],
             "Message": "Creating automated snapshot",
diff --git a/moto/secretsmanager/models.py b/moto/secretsmanager/models.py
index 67e6eb3d5..5ff91af0c 100644
--- a/moto/secretsmanager/models.py
+++ b/moto/secretsmanager/models.py
@@ -132,6 +132,13 @@ class FakeSecret:
             replica_regions or [], force_overwrite=force_overwrite
         )
 
+    @property
+    def owning_service(self) -> Optional[str]:
+        for tag in self.tags or []:
+            if tag["Key"] == "aws:secretsmanager:owningService":
+                return tag["Value"]
+        return None
+
     def create_replicas(
         self, replica_regions: List[Dict[str, str]], force_overwrite: bool
     ) -> List["ReplicaSecret"]:
@@ -237,13 +244,16 @@ class FakeSecret:
         dct: Dict[str, Any] = {
             "ARN": self.arn,
             "Name": self.name,
-            "KmsKeyId": self.kms_key_id,
             "LastChangedDate": self.last_changed_date,
             "LastAccessedDate": None,
             "NextRotationDate": self.next_rotation_date,
             "DeletedDate": self.deleted_date,
             "CreatedDate": self.created_date,
         }
+        if self.kms_key_id != SecretsManagerBackend.DEFAULT_KMS_KEY_ALIAS:
+            dct["KmsKeyId"] = self.kms_key_id
+        if self.owning_service is not None:
+            dct["OwningService"] = self.owning_service
         if self.tags is not None:
             dct["Tags"] = self.tags
         if self.description:
@@ -296,7 +306,7 @@ class ReplicaSecret:
         self.has_replica = status is None
         self.config = {
             "Region": self.region,
-            "KmsKeyId": "alias/aws/secretsmanager",
+            "KmsKeyId": SecretsManagerBackend.DEFAULT_KMS_KEY_ALIAS,
             "Status": self.status,
             "StatusMessage": self.message,
         }
@@ -365,6 +375,8 @@ class SecretsStore(Dict[str, Union[FakeSecret, ReplicaSecret]]):
 
 
 class SecretsManagerBackend(BaseBackend):
+    DEFAULT_KMS_KEY_ALIAS = "alias/aws/secretsmanager"
+
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.secrets = SecretsStore()
@@ -660,6 +672,43 @@ class SecretsManagerBackend(BaseBackend):
 
         return secret, new_version
 
+    def create_managed_secret(
+        self,
+        service_name: str,
+        secret_id: str,
+        secret_string: Optional[str] = None,
+        secret_binary: Optional[str] = None,
+        description: Optional[str] = None,
+        tags: Optional[List[Dict[str, str]]] = None,
+        kms_key_id: Optional[str] = None,
+        version_id: Optional[str] = None,
+        replica_regions: Optional[List[Dict[str, str]]] = None,
+        force_overwrite: bool = False,
+    ) -> FakeSecret:
+        """Create an AWS managed secret for the specified service name."""
+        if kms_key_id is None:
+            kms_key_id = self.DEFAULT_KMS_KEY_ALIAS
+        managed_tag = {
+            "Key": "aws:secretsmanager:owningService",
+            "Value": service_name,
+        }
+        if tags is None:
+            tags = [managed_tag]
+        else:
+            tags.append(managed_tag)
+        secret, _ = self._add_secret(
+            secret_id,
+            secret_string=secret_string,
+            secret_binary=secret_binary,
+            description=description,
+            tags=tags,
+            kms_key_id=kms_key_id,
+            version_id=version_id,
+            replica_regions=replica_regions,
+            force_overwrite=force_overwrite,
+        )
+        return secret
+
     def put_secret_value(
         self,
         secret_id: str,


Test patch:
diff --git a/tests/test_events/test_events.py b/tests/test_events/test_events.py
index 17bd2960b..aeb6fa5a1 100644
--- a/tests/test_events/test_events.py
+++ b/tests/test_events/test_events.py
@@ -2418,6 +2418,7 @@ def test_kms_key_is_created(auth_type, auth_parameters, with_headers):
         assert secret["Tags"] == [
             {"Key": "aws:secretsmanager:owningService", "Value": "events"}
         ]
+        assert secret["OwningService"] == "events"
 
         if auth_type == "BASIC":
             expected_secret = {"username": "un", "password": "pw"}
diff --git a/tests/test_rds/test_integrations.py b/tests/test_rds/test_integrations.py
new file mode 100644
index 000000000..199105045
--- /dev/null
+++ b/tests/test_rds/test_integrations.py
@@ -0,0 +1,366 @@
+"""RDS tests covering integrations with other service backends."""
+
+import uuid
+
+import boto3
+import pytest
+from botocore import waiter, xform_name
+from botocore.exceptions import ClientError
+
+from tests import aws_verified
+
+from . import DEFAULT_REGION
+
+
+@aws_verified
+@pytest.mark.aws_verified
+def test_db_cluster_managed_master_user_password_lifecycle():
+    # Clients
+    kms = boto3.client("kms", region_name=DEFAULT_REGION)
+    rds = boto3.client("rds", region_name=DEFAULT_REGION)
+    secretsmanager = boto3.client("secretsmanager", region_name=DEFAULT_REGION)
+
+    # Waiters
+    db_cluster_modifications_complete = get_custom_waiter(
+        "db_cluster_modifications_complete", rds
+    )
+    db_cluster_available = rds.get_waiter("db_cluster_available")
+    db_cluster_deleted = rds.get_waiter("db_cluster_deleted")
+    master_user_secret_available = get_custom_waiter(
+        "db_cluster_master_user_secret_available", rds
+    )
+    master_user_secret_deleted = get_custom_waiter("secret_deleted", secretsmanager)
+
+    # KMS keys
+    resp = kms.describe_key(KeyId="alias/aws/secretsmanager")
+    default_kms_key_arn = resp["KeyMetadata"]["Arn"]
+    resp = kms.create_key(KeyUsage="ENCRYPT_DECRYPT", KeySpec="SYMMETRIC_DEFAULT")
+    custom_kms_key_arn = resp["KeyMetadata"]["Arn"]
+    custom_kms_key_id = resp["KeyMetadata"]["KeyId"]
+
+    # Create DBCluster with managed master user password
+    db_cluster_identifier = "test-" + str(uuid.uuid4())
+    resp = rds.create_db_cluster(
+        DBClusterIdentifier=db_cluster_identifier,
+        Engine="aurora-postgresql",
+        MasterUsername="root",
+        ManageMasterUserPassword=True,
+    )
+    db_cluster = resp["DBCluster"]
+    master_user_secret = db_cluster["MasterUserSecret"]
+    assert master_user_secret["KmsKeyId"] == default_kms_key_arn
+    assert master_user_secret["SecretStatus"] == "creating"
+    # Check the secret in SecretsManager
+    secret_arn = master_user_secret["SecretArn"]
+    master_user_secret_available.wait(DBClusterIdentifier=db_cluster_identifier)
+    secret = secretsmanager.describe_secret(SecretId=secret_arn)
+    assert str(secret["Name"]).startswith("rds!cluster")
+    assert secret["OwningService"] == "rds"
+    # Check that RDS sees the secret as active
+    db_cluster_available.wait(DBClusterIdentifier=db_cluster_identifier)
+    resp = rds.describe_db_clusters(DBClusterIdentifier=db_cluster_identifier)
+    db_cluster = resp["DBClusters"][0]
+    master_user_secret = db_cluster["MasterUserSecret"]
+    assert master_user_secret["SecretStatus"] == "active"
+
+    # Disable password management
+    resp = rds.modify_db_cluster(
+        DBClusterIdentifier=db_cluster_identifier,
+        ManageMasterUserPassword=False,
+        MasterUserPassword="Non-Managed-Pa55word!",
+        ApplyImmediately=True,
+    )
+    db_cluster = resp["DBCluster"]
+    # TODO: This passes on AWS but is not implemented yet in moto
+    # assert "MasterUserPassword" in db_cluster["PendingModifiedValues"]
+    db_cluster_modifications_complete.wait(DBClusterIdentifier=db_cluster_identifier)
+    resp = rds.describe_db_clusters(DBClusterIdentifier=db_cluster_identifier)
+    db_cluster = resp["DBClusters"][0]
+    assert "MasterUserSecret" not in db_cluster
+    # Secret should be deleted
+    master_user_secret_deleted.wait(SecretId=secret_arn)
+    with pytest.raises(ClientError) as exc:
+        secretsmanager.describe_secret(SecretId=secret_arn)
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
+
+    # Re-enable password management, this time with a custom KMS key
+    resp = rds.modify_db_cluster(
+        DBClusterIdentifier=db_cluster_identifier,
+        ManageMasterUserPassword=True,
+        MasterUserSecretKmsKeyId=custom_kms_key_arn,
+        ApplyImmediately=True,
+    )
+    db_cluster = resp["DBCluster"]
+    # TODO: This passes on AWS but is not implemented yet in moto
+    # assert "MasterUserPassword" in db_cluster["PendingModifiedValues"]
+    master_user_secret = db_cluster["MasterUserSecret"]
+    assert master_user_secret["KmsKeyId"] == custom_kms_key_arn
+    assert master_user_secret["SecretStatus"] == "creating"
+    # Check the secret in SecretsManager
+    secret_arn = master_user_secret["SecretArn"]
+    master_user_secret_available.wait(DBClusterIdentifier=db_cluster_identifier)
+    secret = secretsmanager.describe_secret(SecretId=secret_arn)
+    assert str(secret["Name"]).startswith("rds!cluster")
+    assert secret["OwningService"] == "rds"
+    # Check that RDS sees the secret as active
+    db_cluster_modifications_complete.wait(DBClusterIdentifier=db_cluster_identifier)
+    resp = rds.describe_db_clusters(DBClusterIdentifier=db_cluster_identifier)
+    db_cluster = resp["DBClusters"][0]
+    master_user_secret = db_cluster["MasterUserSecret"]
+    assert master_user_secret["SecretStatus"] == "active"
+
+    # Rotate managed password
+    resp = rds.modify_db_cluster(
+        DBClusterIdentifier=db_cluster_identifier,
+        RotateMasterUserPassword=True,
+        ApplyImmediately=True,
+    )
+    db_cluster = resp["DBCluster"]
+    master_user_secret = db_cluster["MasterUserSecret"]
+    assert master_user_secret["SecretStatus"] == "rotating"
+    # Check that RDS sees the secret as active after rotation
+    master_user_secret_available.wait(DBClusterIdentifier=db_cluster_identifier)
+    resp = rds.describe_db_clusters(DBClusterIdentifier=db_cluster_identifier)
+    db_cluster = resp["DBClusters"][0]
+    master_user_secret = db_cluster["MasterUserSecret"]
+    assert master_user_secret["SecretStatus"] == "active"
+
+    # Delete the cluster
+    resp = rds.delete_db_cluster(
+        DBClusterIdentifier=db_cluster_identifier,
+        SkipFinalSnapshot=True,
+        DeleteAutomatedBackups=True,
+    )
+    db_cluster = resp["DBCluster"]
+    assert "MasterUserSecret" in db_cluster
+    # Deleting the cluster should delete the secret
+    db_cluster_deleted.wait(DBClusterIdentifier=db_cluster_identifier)
+    master_user_secret_deleted.wait(SecretId=secret_arn)
+    with pytest.raises(ClientError) as exc:
+        secretsmanager.describe_secret(SecretId=secret_arn)
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
+    # Delete the custom KMS key we created
+    kms.schedule_key_deletion(KeyId=custom_kms_key_id, PendingWindowInDays=7)
+
+
+@aws_verified
+@pytest.mark.aws_verified
+def test_db_instance_managed_master_user_password_lifecycle():
+    # Clients
+    kms = boto3.client("kms", region_name=DEFAULT_REGION)
+    rds = boto3.client("rds", region_name=DEFAULT_REGION)
+    secretsmanager = boto3.client("secretsmanager", region_name=DEFAULT_REGION)
+
+    # Waiters
+    db_instance_modifications_complete = get_custom_waiter(
+        "db_instance_modifications_complete", rds
+    )
+    db_instance_available = rds.get_waiter("db_instance_available")
+    db_instance_deleted = rds.get_waiter("db_instance_deleted")
+    master_user_secret_available = get_custom_waiter(
+        "db_instance_master_user_secret_available", rds
+    )
+    master_user_secret_deleted = get_custom_waiter("secret_deleted", secretsmanager)
+
+    # KMS keys
+    resp = kms.describe_key(KeyId="alias/aws/secretsmanager")
+    default_kms_key_arn = resp["KeyMetadata"]["Arn"]
+    resp = kms.create_key(KeyUsage="ENCRYPT_DECRYPT", KeySpec="SYMMETRIC_DEFAULT")
+    custom_kms_key_arn = resp["KeyMetadata"]["Arn"]
+    custom_kms_key_id = resp["KeyMetadata"]["KeyId"]
+
+    # Create DBInstance with managed master user password
+    db_instance_identifier = "test-" + str(uuid.uuid4())
+    resp = rds.create_db_instance(
+        AllocatedStorage=200,
+        DBInstanceIdentifier=db_instance_identifier,
+        DBInstanceClass="db.m7g.large",
+        Engine="postgres",
+        MasterUsername="root",
+        ManageMasterUserPassword=True,
+    )
+    db_instance = resp["DBInstance"]
+    master_user_secret = db_instance["MasterUserSecret"]
+    assert master_user_secret["KmsKeyId"] == default_kms_key_arn
+    assert master_user_secret["SecretStatus"] == "creating"
+    # Check the secret in SecretsManager
+    secret_arn = master_user_secret["SecretArn"]
+    master_user_secret_available.wait(DBInstanceIdentifier=db_instance_identifier)
+    secret = secretsmanager.describe_secret(SecretId=secret_arn)
+    assert str(secret["Name"]).startswith("rds!db")
+    assert secret["OwningService"] == "rds"
+    # Check that RDS sees the secret as active
+    db_instance_available.wait(DBInstanceIdentifier=db_instance_identifier)
+    resp = rds.describe_db_instances(DBInstanceIdentifier=db_instance_identifier)
+    db_instance = resp["DBInstances"][0]
+    master_user_secret = db_instance["MasterUserSecret"]
+    assert master_user_secret["SecretStatus"] == "active"
+
+    # Disable password management
+    resp = rds.modify_db_instance(
+        DBInstanceIdentifier=db_instance_identifier,
+        ManageMasterUserPassword=False,
+        MasterUserPassword="Non-Managed-Pa55word!",
+        ApplyImmediately=True,
+    )
+    db_instance = resp["DBInstance"]
+    # TODO: This passes on AWS but is not implemented yet in moto
+    # assert "MasterUserPassword" in db_cluster["PendingModifiedValues"]
+    db_instance_modifications_complete.wait(DBInstanceIdentifier=db_instance_identifier)
+    resp = rds.describe_db_instances(DBInstanceIdentifier=db_instance_identifier)
+    db_instance = resp["DBInstances"][0]
+    assert "MasterUserSecret" not in db_instance
+    # Secret should be deleted
+    master_user_secret_deleted.wait(SecretId=secret_arn)
+    with pytest.raises(ClientError) as exc:
+        secretsmanager.describe_secret(SecretId=secret_arn)
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
+
+    # Re-enable password management, this time with a custom KMS key
+    resp = rds.modify_db_instance(
+        DBInstanceIdentifier=db_instance_identifier,
+        ManageMasterUserPassword=True,
+        MasterUserSecretKmsKeyId=custom_kms_key_arn,
+        ApplyImmediately=True,
+    )
+    db_instance = resp["DBInstance"]
+    # TODO: This passes on AWS but not implemented yet in moto
+    # assert "MasterUserPassword" in db_cluster["PendingModifiedValues"]
+    master_user_secret = db_instance["MasterUserSecret"]
+    assert master_user_secret["KmsKeyId"] == custom_kms_key_arn
+    assert master_user_secret["SecretStatus"] == "creating"
+    # Check the secret in SecretsManager
+    secret_arn = master_user_secret["SecretArn"]
+    master_user_secret_available.wait(DBInstanceIdentifier=db_instance_identifier)
+    secret = secretsmanager.describe_secret(SecretId=secret_arn)
+    assert str(secret["Name"]).startswith("rds!db")
+    assert secret["OwningService"] == "rds"
+    # Check that RDS sees the secret as active
+    db_instance_modifications_complete.wait(DBInstanceIdentifier=db_instance_identifier)
+    resp = rds.describe_db_instances(DBInstanceIdentifier=db_instance_identifier)
+    db_instance = resp["DBInstances"][0]
+    master_user_secret = db_instance["MasterUserSecret"]
+    assert master_user_secret["SecretStatus"] == "active"
+
+    # Rotate managed password
+    resp = rds.modify_db_instance(
+        DBInstanceIdentifier=db_instance_identifier,
+        RotateMasterUserPassword=True,
+        ApplyImmediately=True,
+    )
+    db_instance = resp["DBInstance"]
+    master_user_secret = db_instance["MasterUserSecret"]
+    assert master_user_secret["SecretStatus"] == "rotating"
+    # Check that RDS sees the secret as active after rotation
+    master_user_secret_available.wait(DBInstanceIdentifier=db_instance_identifier)
+    resp = rds.describe_db_instances(DBInstanceIdentifier=db_instance_identifier)
+    db_instance = resp["DBInstances"][0]
+    master_user_secret = db_instance["MasterUserSecret"]
+    assert master_user_secret["SecretStatus"] == "active"
+
+    # Delete the instance
+    resp = rds.delete_db_instance(
+        DBInstanceIdentifier=db_instance_identifier,
+        SkipFinalSnapshot=True,
+        DeleteAutomatedBackups=True,
+    )
+    db_instance = resp["DBInstance"]
+    assert "MasterUserSecret" in db_instance
+    # Deleting the instance should delete the secret
+    db_instance_deleted.wait(DBInstanceIdentifier=db_instance_identifier)
+    master_user_secret_deleted.wait(SecretId=secret_arn)
+    with pytest.raises(ClientError) as exc:
+        secretsmanager.describe_secret(SecretId=secret_arn)
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
+    # Delete the custom KMS key we created
+    kms.schedule_key_deletion(KeyId=custom_kms_key_id, PendingWindowInDays=7)
+
+
+def get_custom_waiter(waiter_name, client):
+    """Some useful waiters that are not present in botocore."""
+    config = {
+        "version": 2,
+        "waiters": {
+            "DBClusterModificationsComplete": {
+                "delay": 15,
+                "operation": "DescribeDBClusters",
+                "maxAttempts": 60,
+                "acceptors": [
+                    {
+                        "state": "success",
+                        "matcher": "path",
+                        "argument": "contains(keys(DBClusters[0]), 'PendingModifiedValues') && length(DBClusters[0].PendingModifiedValues) == `0`",
+                        "expected": True,
+                    },
+                    {
+                        "state": "success",
+                        "matcher": "path",
+                        "argument": "contains(keys(DBClusters[0]), 'PendingModifiedValues')",
+                        "expected": False,
+                    },
+                ],
+            },
+            "DBClusterMasterUserSecretAvailable": {
+                "delay": 15,
+                "operation": "DescribeDBClusters",
+                "maxAttempts": 60,
+                "acceptors": [
+                    {
+                        "expected": "active",
+                        "matcher": "pathAll",
+                        "state": "success",
+                        "argument": "DBClusters[].MasterUserSecret.SecretStatus",
+                    },
+                ],
+            },
+            "DBInstanceModificationsComplete": {
+                "delay": 15,
+                "operation": "DescribeDBInstances",
+                "maxAttempts": 60,
+                "acceptors": [
+                    {
+                        "state": "success",
+                        "matcher": "path",
+                        "argument": "contains(keys(DBInstances[0]), 'PendingModifiedValues') && length(DBInstances[0].PendingModifiedValues) == `0`",
+                        "expected": True,
+                    },
+                    {
+                        "state": "success",
+                        "matcher": "path",
+                        "argument": "contains(keys(DBInstances[0]), 'PendingModifiedValues')",
+                        "expected": False,
+                    },
+                ],
+            },
+            "DBInstanceMasterUserSecretAvailable": {
+                "delay": 15,
+                "operation": "DescribeDBInstances",
+                "maxAttempts": 60,
+                "acceptors": [
+                    {
+                        "expected": "active",
+                        "matcher": "pathAll",
+                        "state": "success",
+                        "argument": "DBInstances[].MasterUserSecret.SecretStatus",
+                    },
+                ],
+            },
+            "SecretDeleted": {
+                "delay": 15,
+                "operation": "DescribeSecret",
+                "maxAttempts": 60,
+                "acceptors": [
+                    {
+                        "expected": "ResourceNotFoundException",
+                        "matcher": "error",
+                        "state": "success",
+                    },
+                ],
+            },
+        },
+    }
+    model = waiter.WaiterModel(config)
+    mapping = {}
+    for name in model.waiter_names:
+        mapping[xform_name(name)] = name
+    return waiter.create_waiter_with_client(mapping[waiter_name], model, client)
diff --git a/tests/test_rds/test_rds.py b/tests/test_rds/test_rds.py
index 02d7b06a8..91e387dfe 100644
--- a/tests/test_rds/test_rds.py
+++ b/tests/test_rds/test_rds.py
@@ -461,8 +461,11 @@ def test_modify_db_instance_manage_master_user_password(
     with_custom_kms_key: bool, client
 ):
     db_id = "db-id"
-
-    custom_kms_key = f"arn:aws:kms:{DEFAULT_REGION}:123456789012:key/abcd1234-56ef-78gh-90ij-klmnopqrstuv"
+    kms = boto3.client("kms", region_name=DEFAULT_REGION)
+    key = kms.create_key(KeyUsage="ENCRYPT_DECRYPT", KeySpec="SYMMETRIC_DEFAULT")[
+        "KeyMetadata"
+    ]
+    custom_kms_key = key["Arn"]
     custom_kms_key_args = (
         {"MasterUserSecretKmsKeyId": custom_kms_key} if with_custom_kms_key else {}
     )
@@ -485,22 +488,25 @@ def test_modify_db_instance_manage_master_user_password(
     assert db_instance.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 str(master_user_secret["SecretArn"]).startswith(
+        f"arn:aws:secretsmanager:{DEFAULT_REGION}:{ACCOUNT_ID}:secret:rds!db"
     )
-    assert master_user_secret["SecretStatus"] == "active"
+    assert master_user_secret["SecretStatus"] == "creating"
     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"
-        )
+        default_kms_key = kms.describe_key(KeyId="alias/aws/secretsmanager")[
+            "KeyMetadata"
+        ]["Arn"]
+        assert master_user_secret["KmsKeyId"] == default_kms_key
     assert len(describe_response["DBInstances"][0]["MasterUserSecret"].keys()) == 3
     assert (
-        modify_response["DBInstance"]["MasterUserSecret"]
-        == describe_response["DBInstances"][0]["MasterUserSecret"]
+        describe_response["DBInstances"][0]["MasterUserSecret"]["SecretStatus"]
+        == "active"
+    )
+    assert (
+        modify_response["DBInstance"]["MasterUserSecret"]["SecretArn"]
+        == describe_response["DBInstances"][0]["MasterUserSecret"]["SecretArn"]
     )
     assert revert_modification_response["DBInstance"].get("MasterUserSecret") is None
 
diff --git a/tests/test_rds/test_rds_clusters.py b/tests/test_rds/test_rds_clusters.py
index ba94a8b30..1a93dab22 100644
--- a/tests/test_rds/test_rds_clusters.py
+++ b/tests/test_rds/test_rds_clusters.py
@@ -158,7 +158,11 @@ def test_modify_db_cluster_new_cluster_identifier(client):
 @mock_aws
 def test_modify_db_cluster_manage_master_user_password(client, with_custom_kms_key):
     cluster_id = "cluster-id"
-    custom_kms_key = f"arn:aws:kms:{DEFAULT_REGION}:123456789012:key/abcd1234-56ef-78gh-90ij-klmnopqrstuv"
+    kms = boto3.client("kms", region_name=DEFAULT_REGION)
+    key = kms.create_key(KeyUsage="ENCRYPT_DECRYPT", KeySpec="SYMMETRIC_DEFAULT")[
+        "KeyMetadata"
+    ]
+    custom_kms_key = key["Arn"]
     custom_kms_key_args = (
         {"MasterUserSecretKmsKeyId": custom_kms_key} if with_custom_kms_key else {}
     )
@@ -188,22 +192,25 @@ def test_modify_db_cluster_manage_master_user_password(client, with_custom_kms_k
     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"]
-        == f"arn:aws:secretsmanager:{DEFAULT_REGION}:123456789012:secret:rds!cluster-id"
+    assert str(master_user_secret["SecretArn"]).startswith(
+        f"arn:aws:secretsmanager:{DEFAULT_REGION}:{ACCOUNT_ID}:secret:rds!cluster"
     )
-    assert master_user_secret["SecretStatus"] == "active"
+    assert master_user_secret["SecretStatus"] == "creating"
     if with_custom_kms_key:
         assert master_user_secret["KmsKeyId"] == custom_kms_key
     else:
-        assert (
-            master_user_secret["KmsKeyId"]
-            == f"arn:aws:kms:{DEFAULT_REGION}:123456789012:key/cluster-id"
-        )
+        default_kms_key = kms.describe_key(KeyId="alias/aws/secretsmanager")[
+            "KeyMetadata"
+        ]["Arn"]
+        assert master_user_secret["KmsKeyId"] == default_kms_key
     assert len(describe_response["DBClusters"][0]["MasterUserSecret"].keys()) == 3
     assert (
-        modify_response["DBCluster"]["MasterUserSecret"]
-        == describe_response["DBClusters"][0]["MasterUserSecret"]
+        describe_response["DBClusters"][0]["MasterUserSecret"]["SecretStatus"]
+        == "active"
+    )
+    assert (
+        modify_response["DBCluster"]["MasterUserSecret"]["SecretArn"]
+        == describe_response["DBClusters"][0]["MasterUserSecret"]["SecretArn"]
     )
     assert revert_modification_response["DBCluster"].get("MasterUserSecret") is None
 
diff --git a/tests/test_secretsmanager/test_secretsmanager.py b/tests/test_secretsmanager/test_secretsmanager.py
index a98e6295f..7b7ee3ce6 100644
--- a/tests/test_secretsmanager/test_secretsmanager.py
+++ b/tests/test_secretsmanager/test_secretsmanager.py
@@ -14,6 +14,7 @@ from freezegun import freeze_time
 
 from moto import mock_aws, settings
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
+from moto.secretsmanager.models import secretsmanager_backends
 from moto.secretsmanager.utils import SecretsManagerSecretIdentifier
 from moto.utilities.id_generator import TAG_KEY_CUSTOM_ID
 from tests import allow_aws_request
@@ -2180,3 +2181,21 @@ def test_create_secret_with_tag_custom_id(set_custom_id):
     )
 
     assert secret["ARN"].split(":")[-1] == f"{secret_name}-{secret_suffix}"
+
+
+@mock_aws
+@pytest.mark.skipif(
+    not settings.TEST_DECORATOR_MODE,
+    reason="Can't modify backend directly if not in decorator mode.",
+)
+def test_aws_managed_secret(set_custom_id):
+    # We have to poke directly at the backend because technically this secret
+    # would be managed by RDS, and wouldn't be able to be created via the
+    # public API (due to the restricted 'aws' tag prefix).
+    backend = secretsmanager_backends[DEFAULT_ACCOUNT_ID]["us-east-1"]
+    secret = backend.create_managed_secret("rds", "secret-managed-by-rds")
+    assert secret.kms_key_id == "alias/aws/secretsmanager"
+    client = boto3.client("secretsmanager", region_name="us-east-1")
+    resp = client.describe_secret(SecretId=secret.arn)
+    assert "KmsKeyId" not in resp
+    assert resp["OwningService"] == "rds"