Repository: getmoto/moto
Pull request #7865 (2024-07-17 23:08:19)

IoT: Set Certificate ID to SHA256 hexdigest of DER.
Previous algorithm was based on the PEM format and produced an incorrect certificate ID.
Includes configurable certificate validity for backwards compatibility.

mcli: This is an spin of https://github.com/getmoto/moto/pull/6858 which was closed.  Make the default to generate the correct cert ID and allow tests to optionally configure to use previous algorithm which supported invalid certificates in the test.
mcli: Fixes https://github.com/getmoto/moto/issues/7866
mcli: > Hi @mcli, thanks for picking this up. The change makes sense to me - but we would have to keep the current (incorrect) behaviour as the default, and make the DER calculation the opt-in behaviour. We don't want to introduce a breaking change for users.

Hi @bblommers , Ok.  Set the default to `False`.  Thanks for the comments and I agree with the approach.  Propose to change the default in the next major release.

Patch:
diff --git a/AUTHORS.md b/AUTHORS.md
index 0c7b081ad..d5de2bbb6 100644
--- a/AUTHORS.md
+++ b/AUTHORS.md
@@ -63,3 +63,4 @@ Moto is written by Steve Pulec with contributions from:
 * [Roman Rader](https://github.com/rrader/)
 * [Bryan Chen](https://github.com/bchen1116)
 * [Jonas Bulik](https://github.com/MrGreenTea)
+* [Rita Cheung](https://github.com/rita-cheung)
diff --git a/docs/docs/configuration/index.rst b/docs/docs/configuration/index.rst
index 106986b90..151b48eba 100644
--- a/docs/docs/configuration/index.rst
+++ b/docs/docs/configuration/index.rst
@@ -23,6 +23,7 @@ If you are using the decorators, some options are configurable within the decora
         },
         "iam": {"load_aws_managed_policies": False},
         "stepfunctions": {"execute_state_machine": True},
+        "iot": {"use_valid_cert": False},
     })
 
 
@@ -57,6 +58,24 @@ Moto comes bundled with all Managed Policies that AWS exposes, which are updated
 
 Set `"iam": {"load_aws_managed_policies": True}` to load these policies for a single test.
 
+IoT Certificates Validity and Certificate ID computation
+-------------------------
+The IoT Client implementation for certificates, by default,
+computes the certificate ID incorrectly based on the
+sha256 hash of the certificate in PEM encoding and does not require a
+valid certificate.
+
+We now support proper certificate ID generation using 
+the sha256 of the DER certificate to ensure correct behavior for tests that
+use valid certificates.  To enable the correct
+certificate ID generation algorithm, set `"iot": {"use_valid_cert": True}`
+
+In the next major release we recommend changing the default to `True`
+which would a breaking change for tests that do not use valid certificates.
+For future compatiblity, update any existing code that does not use real
+certificates to explicitly set `use_valid_cert` to `False`
+
+
 Configuring MotoServer
 ----------------------
 The following options can also be configured when running the MotoServer:
diff --git a/moto/core/config.py b/moto/core/config.py
index 5958c44fc..1b6cfa72b 100644
--- a/moto/core/config.py
+++ b/moto/core/config.py
@@ -25,6 +25,10 @@ class _sfn_config(TypedDict, total=False):
     execute_state_machine: bool
 
 
+class _iot_config(TypedDict, total=False):
+    use_valid_cert: bool
+
+
 DefaultConfig = TypedDict(
     "DefaultConfig",
     {
@@ -33,6 +37,7 @@ DefaultConfig = TypedDict(
         "lambda": _docker_config,
         "iam": _iam_config,
         "stepfunctions": _sfn_config,
+        "iot": _iot_config,
     },
     total=False,
 )
@@ -47,6 +52,7 @@ default_user_config: DefaultConfig = {
     },
     "iam": {"load_aws_managed_policies": False},
     "stepfunctions": {"execute_state_machine": False},
+    "iot": {"use_valid_cert": False},
 }
 
 
diff --git a/moto/core/utils.py b/moto/core/utils.py
index 921ccfd3e..9a25be112 100644
--- a/moto/core/utils.py
+++ b/moto/core/utils.py
@@ -1,7 +1,7 @@
 import datetime
 import inspect
 import re
-from gzip import decompress
+from gzip import compress, decompress
 from typing import Any, Callable, Dict, List, Optional, Tuple
 from urllib.parse import ParseResult, urlparse
 
@@ -403,6 +403,10 @@ def gzip_decompress(body: bytes) -> bytes:
     return decompress(body)
 
 
+def gzip_compress(body: bytes) -> bytes:
+    return compress(body)
+
+
 ISO_REGION_DOMAINS = {
     "iso": "c2s.ic.gov",
     "isob": "sc2s.sgov.gov",
diff --git a/moto/dynamodb/models/__init__.py b/moto/dynamodb/models/__init__.py
index 9b114cbe3..ab6997bb2 100644
--- a/moto/dynamodb/models/__init__.py
+++ b/moto/dynamodb/models/__init__.py
@@ -37,6 +37,7 @@ from moto.dynamodb.models.table import (
     RestoredTable,
     Table,
 )
+from moto.dynamodb.models.table_export import TableExport
 from moto.dynamodb.models.table_import import TableImport
 from moto.dynamodb.parsing import partiql
 from moto.dynamodb.parsing.executors import UpdateExpressionExecutor
@@ -53,6 +54,7 @@ class DynamoDBBackend(BaseBackend):
         self.tables: Dict[str, Table] = OrderedDict()
         self.backups: Dict[str, Backup] = OrderedDict()
         self.table_imports: Dict[str, TableImport] = {}
+        self.table_exports: Dict[str, TableExport] = {}
 
     @staticmethod
     def default_vpc_endpoint_service(
@@ -999,5 +1001,41 @@ class DynamoDBBackend(BaseBackend):
     def describe_import(self, import_arn: str) -> TableImport:
         return self.table_imports[import_arn]
 
+    def export_table(
+        self,
+        s3_bucket: str,
+        s3_prefix: str,
+        table_arn: str,
+        export_format: str,
+        export_type: str,
+        s3_bucket_owner: str,
+    ) -> TableExport:
+        """Only ExportFormat=DYNAMODB_JSON is supported so far.
+        Only exports one file following DYNAMODB_JSON format to the s3 location. Other files aren't created.
+        Incremental export is also not supported.
+        """
+        table_export = TableExport(
+            s3_bucket=s3_bucket,
+            s3_prefix=s3_prefix,
+            region_name=self.region_name,
+            account_id=s3_bucket_owner if s3_bucket_owner else self.account_id,
+            table_arn=table_arn,
+            export_format=export_format,
+            export_type=export_type,
+        )
+        self.table_exports[table_export.arn] = table_export
+        table_export.start()
+        return table_export
+
+    def describe_export(self, export_arn: str) -> TableExport:
+        return self.table_exports[export_arn]
+
+    def list_exports(self, table_arn: str) -> List[TableExport]:
+        exports = []
+        for export_arn in self.table_exports:
+            if self.table_exports[export_arn].table_arn == table_arn:
+                exports.append(self.table_exports[export_arn])
+        return exports
+
 
 dynamodb_backends = BackendDict(DynamoDBBackend, "dynamodb")
diff --git a/moto/dynamodb/models/table_export.py b/moto/dynamodb/models/table_export.py
new file mode 100644
index 000000000..f596e0d6c
--- /dev/null
+++ b/moto/dynamodb/models/table_export.py
@@ -0,0 +1,112 @@
+import json
+from threading import Thread
+from typing import TYPE_CHECKING, Any, Dict, Optional
+from uuid import uuid4
+
+from moto.core.utils import gzip_compress
+from moto.utilities.utils import get_partition
+
+if TYPE_CHECKING:
+    from moto.dynamodb.models import Table
+    from moto.s3.models import S3Backend
+
+
+class TableExport(Thread):
+    def __init__(
+        self,
+        s3_bucket: str,
+        s3_prefix: str,
+        region_name: str,
+        account_id: str,
+        table_arn: str,
+        export_format: str,
+        export_type: str,
+    ):
+        super().__init__()
+        self.partition = get_partition(region_name)
+        self.table_arn = table_arn
+        self.arn = f"arn:{self.partition}:dynamodb:{region_name}:{account_id}:table/{table_arn}/import/{str(uuid4()).replace('-', '')}"
+        self.s3_bucket = s3_bucket
+        self.s3_prefix = s3_prefix
+        self.status = "IN_PROGRESS"
+        self.export_format = export_format
+        self.export_type = export_type
+        self.account_id = account_id
+        self.region_name = region_name
+
+        self.failure_code: Optional[str] = None
+        self.failure_message: Optional[str] = None
+        self.table_name: str = ""
+        self.item_count = 0
+        self.processed_bytes = 0
+        self.error_count = 0
+
+    def run(self) -> None:
+        from moto.dynamodb.models import dynamodb_backends
+
+        dynamodb_backend = dynamodb_backends[self.account_id][self.region_name]
+        try:
+            from moto.s3.models import s3_backends
+
+            s3_backend = s3_backends[self.account_id][self.partition]
+            s3_backend.buckets[self.s3_bucket]
+
+        except KeyError:
+            self.status = "FAILED"
+            self.failure_code = "S3NoSuchBucket"
+            self.failure_message = "The specified bucket does not exist"
+            return
+
+        for key in dynamodb_backend.tables:
+            if dynamodb_backend.tables[key].table_arn == self.table_arn:
+                self.table_name = key
+        if not self.table_name:
+            self.status = "FAILED"
+            self.failure_code = "DynamoDBTableNotFound"
+            self.failure_message = "The specified table does not exist"
+            return
+        table = dynamodb_backend.tables[self.table_name]
+        if (
+            table.continuous_backups["PointInTimeRecoveryDescription"][
+                "PointInTimeRecoveryStatus"
+            ]
+            != "ENABLED"
+        ):
+            self.status = "FAILED"
+            self.failure_code = "PointInTimeRecoveryUnavailable"
+            self.failure_message = "Point in time recovery not enabled for table"
+            return
+        try:
+            self._backup_to_s3_file(s3_backend, table)
+        except Exception as e:
+            self.status = "FAILED"
+            self.failure_code = "UNKNOWN"
+            self.failure_message = str(e)
+
+    def _backup_to_s3_file(self, s3_backend: "S3Backend", table: "Table") -> None:
+        backup = []
+        for item in table.all_items():
+            json_item = item.to_json()
+            backup.append(json_item)
+            self.processed_bytes += len(json_item)
+        self.item_count = len(backup)
+        content = gzip_compress(json.dumps(backup).encode("utf-8"))
+        s3_backend.put_object(
+            bucket_name=self.s3_bucket,
+            key_name=f"{self.s3_prefix}/AWSDynamoDB/{str(uuid4())}/data/{str(uuid4())}.gz",
+            value=content,
+        )
+
+        self.status = "COMPLETED" if self.error_count == 0 else "FAILED"
+
+    def response(self) -> Dict[str, Any]:
+        return {
+            "ExportArn": self.arn,
+            "ExportStatus": self.status,
+            "FailureCode": self.failure_code,
+            "FailureMessage": self.failure_message,
+            "ExportFormat": self.export_format,
+            "ExportType": self.export_type,
+            "ItemCount": self.item_count,
+            "BilledSizeBytes": self.processed_bytes,
+        }
diff --git a/moto/dynamodb/responses.py b/moto/dynamodb/responses.py
index b8a581e02..d77ed7cdf 100644
--- a/moto/dynamodb/responses.py
+++ b/moto/dynamodb/responses.py
@@ -1227,3 +1227,41 @@ class DynamoHandler(BaseResponse):
         import_arn = self.body["ImportArn"]
         import_table = self.dynamodb_backend.describe_import(import_arn)
         return json.dumps({"ImportTableDescription": import_table.response()})
+
+    def export_table_to_point_in_time(self) -> str:
+        table_arn = self.body["TableArn"]
+        s3_bucket = self.body["S3Bucket"]
+        s3_prefix = self.body["S3Prefix"]
+        export_type = self.body.get("ExportType", "FULL_EXPORT")
+        export_format = self.body.get("ExportFormat", "DYNAMO_JSON")
+        s3_bucket_owner = self.body.get("S3BucketOwner")
+
+        export_table = self.dynamodb_backend.export_table(
+            s3_bucket=s3_bucket,
+            s3_prefix=s3_prefix,
+            table_arn=table_arn,
+            export_type=export_type,
+            export_format=export_format,
+            s3_bucket_owner=s3_bucket_owner,
+        )
+
+        return dynamo_json_dump({"ExportDescription": export_table.response()})
+
+    def describe_export(self) -> str:
+        export_arn = self.body["ExportArn"]
+        export_table = self.dynamodb_backend.describe_export(export_arn)
+        return json.dumps({"ExportDescription": export_table.response()})
+
+    def list_exports(self) -> str:
+        table_arn = self.body["TableArn"]
+        exports = self.dynamodb_backend.list_exports(table_arn)
+        response = []
+        for export_table in exports:
+            response.append(
+                {
+                    "ExportArn": export_table.arn,
+                    "ExportStatus": export_table.status,
+                    "ExportType": export_table.export_type,
+                }
+            )
+        return json.dumps({"ExportSummaries": response})
diff --git a/moto/ec2/resources/instance_type_offerings/availability-zone-id/ap-southeast-3.json b/moto/ec2/resources/instance_type_offerings/availability-zone-id/ap-southeast-3.json
index e21fb63ac..66ed5ce7f 100644
--- a/moto/ec2/resources/instance_type_offerings/availability-zone-id/ap-southeast-3.json
+++ b/moto/ec2/resources/instance_type_offerings/availability-zone-id/ap-southeast-3.json
@@ -243,6 +243,30 @@
   "InstanceType": "c6in.xlarge",
   "Location": "apse3-az1"
  },
+ {
+  "InstanceType": "d3en.12xlarge",
+  "Location": "apse3-az1"
+ },
+ {
+  "InstanceType": "d3en.2xlarge",
+  "Location": "apse3-az1"
+ },
+ {
+  "InstanceType": "d3en.4xlarge",
+  "Location": "apse3-az1"
+ },
+ {
+  "InstanceType": "d3en.6xlarge",
+  "Location": "apse3-az1"
+ },
+ {
+  "InstanceType": "d3en.8xlarge",
+  "Location": "apse3-az1"
+ },
+ {
+  "InstanceType": "d3en.xlarge",
+  "Location": "apse3-az1"
+ },
  {
   "InstanceType": "g5.12xlarge",
   "Location": "apse3-az1"
@@ -1099,6 +1123,30 @@
   "InstanceType": "c6in.xlarge",
   "Location": "apse3-az2"
  },
+ {
+  "InstanceType": "d3en.12xlarge",
+  "Location": "apse3-az2"
+ },
+ {
+  "InstanceType": "d3en.2xlarge",
+  "Location": "apse3-az2"
+ },
+ {
+  "InstanceType": "d3en.4xlarge",
+  "Location": "apse3-az2"
+ },
+ {
+  "InstanceType": "d3en.6xlarge",
+  "Location": "apse3-az2"
+ },
+ {
+  "InstanceType": "d3en.8xlarge",
+  "Location": "apse3-az2"
+ },
+ {
+  "InstanceType": "d3en.xlarge",
+  "Location": "apse3-az2"
+ },
  {
   "InstanceType": "i3.16xlarge",
   "Location": "apse3-az2"
diff --git a/moto/ec2/resources/instance_type_offerings/availability-zone-id/eu-central-1.json b/moto/ec2/resources/instance_type_offerings/availability-zone-id/eu-central-1.json
index 73e6a54c9..f26218770 100644
--- a/moto/ec2/resources/instance_type_offerings/availability-zone-id/eu-central-1.json
+++ b/moto/ec2/resources/instance_type_offerings/availability-zone-id/eu-central-1.json
@@ -2507,6 +2507,10 @@
   "InstanceType": "u-12tb1.112xlarge",
   "Location": "euc1-az1"
  },
+ {
+  "InstanceType": "u-18tb1.112xlarge",
+  "Location": "euc1-az1"
+ },
  {
   "InstanceType": "u-3tb1.56xlarge",
   "Location": "euc1-az1"
diff --git a/moto/ec2/resources/instance_type_offerings/availability-zone/ap-southeast-3.json b/moto/ec2/resources/instance_type_offerings/availability-zone/ap-southeast-3.json
index f8714d143..2f92444e9 100644
--- a/moto/ec2/resources/instance_type_offerings/availability-zone/ap-southeast-3.json
+++ b/moto/ec2/resources/instance_type_offerings/availability-zone/ap-southeast-3.json
@@ -243,6 +243,30 @@
   "InstanceType": "c6in.xlarge",
   "Location": "ap-southeast-3a"
  },
+ {
+  "InstanceType": "d3en.12xlarge",
+  "Location": "ap-southeast-3a"
+ },
+ {
+  "InstanceType": "d3en.2xlarge",
+  "Location": "ap-southeast-3a"
+ },
+ {
+  "InstanceType": "d3en.4xlarge",
+  "Location": "ap-southeast-3a"
+ },
+ {
+  "InstanceType": "d3en.6xlarge",
+  "Location": "ap-southeast-3a"
+ },
+ {
+  "InstanceType": "d3en.8xlarge",
+  "Location": "ap-southeast-3a"
+ },
+ {
+  "InstanceType": "d3en.xlarge",
+  "Location": "ap-southeast-3a"
+ },
  {
   "InstanceType": "g5.12xlarge",
   "Location": "ap-southeast-3a"
@@ -1099,6 +1123,30 @@
   "InstanceType": "c6in.xlarge",
   "Location": "ap-southeast-3b"
  },
+ {
+  "InstanceType": "d3en.12xlarge",
+  "Location": "ap-southeast-3b"
+ },
+ {
+  "InstanceType": "d3en.2xlarge",
+  "Location": "ap-southeast-3b"
+ },
+ {
+  "InstanceType": "d3en.4xlarge",
+  "Location": "ap-southeast-3b"
+ },
+ {
+  "InstanceType": "d3en.6xlarge",
+  "Location": "ap-southeast-3b"
+ },
+ {
+  "InstanceType": "d3en.8xlarge",
+  "Location": "ap-southeast-3b"
+ },
+ {
+  "InstanceType": "d3en.xlarge",
+  "Location": "ap-southeast-3b"
+ },
  {
   "InstanceType": "i3.16xlarge",
   "Location": "ap-southeast-3b"
diff --git a/moto/ec2/resources/instance_type_offerings/availability-zone/eu-central-1.json b/moto/ec2/resources/instance_type_offerings/availability-zone/eu-central-1.json
index e8817fe03..6f48996ff 100644
--- a/moto/ec2/resources/instance_type_offerings/availability-zone/eu-central-1.json
+++ b/moto/ec2/resources/instance_type_offerings/availability-zone/eu-central-1.json
@@ -7839,6 +7839,10 @@
   "InstanceType": "u-12tb1.112xlarge",
   "Location": "eu-central-1c"
  },
+ {
+  "InstanceType": "u-18tb1.112xlarge",
+  "Location": "eu-central-1c"
+ },
  {
   "InstanceType": "u-3tb1.56xlarge",
   "Location": "eu-central-1c"
diff --git a/moto/ec2/resources/instance_type_offerings/region/ap-southeast-3.json b/moto/ec2/resources/instance_type_offerings/region/ap-southeast-3.json
index 904617b58..24081d606 100644
--- a/moto/ec2/resources/instance_type_offerings/region/ap-southeast-3.json
+++ b/moto/ec2/resources/instance_type_offerings/region/ap-southeast-3.json
@@ -243,6 +243,30 @@
   "InstanceType": "c6in.xlarge",
   "Location": "ap-southeast-3"
  },
+ {
+  "InstanceType": "d3en.12xlarge",
+  "Location": "ap-southeast-3"
+ },
+ {
+  "InstanceType": "d3en.2xlarge",
+  "Location": "ap-southeast-3"
+ },
+ {
+  "InstanceType": "d3en.4xlarge",
+  "Location": "ap-southeast-3"
+ },
+ {
+  "InstanceType": "d3en.6xlarge",
+  "Location": "ap-southeast-3"
+ },
+ {
+  "InstanceType": "d3en.8xlarge",
+  "Location": "ap-southeast-3"
+ },
+ {
+  "InstanceType": "d3en.xlarge",
+  "Location": "ap-southeast-3"
+ },
  {
   "InstanceType": "g5.12xlarge",
   "Location": "ap-southeast-3"
diff --git a/moto/ec2/resources/instance_type_offerings/region/eu-central-1.json b/moto/ec2/resources/instance_type_offerings/region/eu-central-1.json
index 5203d58b9..a3e03c2d5 100644
--- a/moto/ec2/resources/instance_type_offerings/region/eu-central-1.json
+++ b/moto/ec2/resources/instance_type_offerings/region/eu-central-1.json
@@ -2747,6 +2747,10 @@
   "InstanceType": "u-12tb1.112xlarge",
   "Location": "eu-central-1"
  },
+ {
+  "InstanceType": "u-18tb1.112xlarge",
+  "Location": "eu-central-1"
+ },
  {
   "InstanceType": "u-3tb1.56xlarge",
   "Location": "eu-central-1"
diff --git a/moto/ec2/resources/instance_types.json b/moto/ec2/resources/instance_types.json
index 4331412c5..24599c7a2 100644
--- a/moto/ec2/resources/instance_types.json
+++ b/moto/ec2/resources/instance_types.json
@@ -7307,7 +7307,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6g.12xlarge",
@@ -7445,7 +7445,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6g.16xlarge",
@@ -7599,7 +7599,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6g.2xlarge",
@@ -7697,7 +7697,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6g.4xlarge",
@@ -7803,7 +7803,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6g.8xlarge",
@@ -7925,7 +7925,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6g.large",
@@ -8017,7 +8017,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6g.medium",
@@ -8186,7 +8186,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6g.xlarge",
@@ -8280,7 +8280,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -8430,7 +8430,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -8596,7 +8596,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -8706,7 +8706,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -8824,7 +8824,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -8958,7 +8958,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -9062,7 +9062,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -9255,7 +9255,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -9361,7 +9361,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6gn.12xlarge",
@@ -9499,7 +9499,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6gn.16xlarge",
@@ -9656,7 +9656,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6gn.2xlarge",
@@ -9754,7 +9754,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6gn.4xlarge",
@@ -9860,7 +9860,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6gn.8xlarge",
@@ -9982,7 +9982,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6gn.large",
@@ -10074,7 +10074,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6gn.medium",
@@ -10165,7 +10165,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c6gn.xlarge",
@@ -14879,7 +14879,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c7g.12xlarge",
@@ -15017,7 +15017,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c7g.16xlarge",
@@ -15174,7 +15174,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c7g.2xlarge",
@@ -15272,7 +15272,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c7g.4xlarge",
@@ -15378,7 +15378,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c7g.8xlarge",
@@ -15500,7 +15500,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c7g.large",
@@ -15592,7 +15592,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c7g.medium",
@@ -15764,7 +15764,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "c7g.xlarge",
@@ -15858,7 +15858,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -16008,7 +16008,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -16177,7 +16177,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -16287,7 +16287,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -16405,7 +16405,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -16539,7 +16539,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -16643,7 +16643,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -16839,7 +16839,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -44220,7 +44220,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m6g.2xlarge",
@@ -44318,7 +44318,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m6g.4xlarge",
@@ -44424,7 +44424,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m6g.8xlarge",
@@ -44546,7 +44546,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m6g.large",
@@ -44638,7 +44638,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m6g.medium",
@@ -44807,7 +44807,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m6g.xlarge",
@@ -45217,7 +45217,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -45327,7 +45327,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -45445,7 +45445,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -45579,7 +45579,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -45683,7 +45683,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -45876,7 +45876,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -52135,7 +52135,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m7g.2xlarge",
@@ -52233,7 +52233,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m7g.4xlarge",
@@ -52339,7 +52339,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m7g.8xlarge",
@@ -52461,7 +52461,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m7g.large",
@@ -52553,7 +52553,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m7g.medium",
@@ -52725,7 +52725,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "m7g.xlarge",
@@ -53138,7 +53138,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -53248,7 +53248,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -53366,7 +53366,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -53500,7 +53500,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -53604,7 +53604,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -53800,7 +53800,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -66707,7 +66707,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "r6g.2xlarge",
@@ -66805,7 +66805,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "r6g.4xlarge",
@@ -67033,7 +67033,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "r6g.large",
@@ -67125,7 +67125,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "r6g.medium",
@@ -67294,7 +67294,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "r6g.xlarge",
@@ -67704,7 +67704,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -67814,7 +67814,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -68066,7 +68066,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -68170,7 +68170,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -68363,7 +68363,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -74622,7 +74622,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "r7g.2xlarge",
@@ -74720,7 +74720,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "r7g.4xlarge",
@@ -74948,7 +74948,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "r7g.large",
@@ -75040,7 +75040,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "r7g.medium",
@@ -75212,7 +75212,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageSupported": false,
   "InstanceType": "r7g.xlarge",
@@ -75625,7 +75625,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -75735,7 +75735,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -75987,7 +75987,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -76091,7 +76091,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
@@ -76287,7 +76287,7 @@
    "NvmeSupport": "required"
   },
   "FreeTierEligible": false,
-  "HibernationSupported": false,
+  "HibernationSupported": true,
   "Hypervisor": "nitro",
   "InstanceStorageInfo": {
    "Disks": [
diff --git a/moto/firehose/models.py b/moto/firehose/models.py
index 232a2b410..2af38c387 100644
--- a/moto/firehose/models.py
+++ b/moto/firehose/models.py
@@ -445,7 +445,7 @@ class FirehoseBackend(BaseBackend):
         # Path prefix pattern: myApp/YYYY/MM/DD/HH/
         # Object name pattern:
         # DeliveryStreamName-DeliveryStreamVersion-YYYY-MM-DD-HH-MM-SS-RandomString
-        prefix = f"{prefix}{'' if prefix.endswith('/') else '/'}"
+        prefix = f"{prefix}{'' if prefix.endswith('/') or prefix == '' else '/'}"
         now = utcnow()
         return (
             f"{prefix}{now.strftime('%Y/%m/%d/%H')}/"
diff --git a/moto/iot/models.py b/moto/iot/models.py
index fe2d0ba9d..b46d53e41 100644
--- a/moto/iot/models.py
+++ b/moto/iot/models.py
@@ -15,6 +15,7 @@ from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.core.utils import utcnow
 from moto.moto_api._internal import mock_random as random
+from moto.settings import iot_use_valid_cert
 from moto.utilities.paginator import paginate
 from moto.utilities.utils import get_partition
 
@@ -192,11 +193,10 @@ class FakeCertificate(BaseModel):
         region_name: str,
         ca_certificate_id: Optional[str] = None,
     ):
-        m = hashlib.sha256()
-        m.update(certificate_pem.encode("utf-8"))
-        self.certificate_id = m.hexdigest()
+        self.certificate_id = self.compute_cert_id(certificate_pem)
         self.arn = f"arn:{get_partition(region_name)}:iot:{region_name}:{account_id}:cert/{self.certificate_id}"
         self.certificate_pem = certificate_pem
+
         self.status = status
 
         self.owner = account_id
@@ -207,6 +207,32 @@ class FakeCertificate(BaseModel):
         self.validity_not_after = time.time() + 86400
         self.ca_certificate_id = ca_certificate_id
 
+    def compute_cert_id(self, certificate_pem: str) -> str:
+        if iot_use_valid_cert():
+            return self.compute_der_cert_id(certificate_pem)
+        else:
+            return self.compute_pem_cert_id(certificate_pem)
+
+    def compute_pem_cert_id(self, certificate_pem: str) -> str:
+        """
+        Original (incorrect) PEM-based hash kept for backwards compatibility
+        with existing tests which may or may not use valid certs. Also useful
+        for simplifying tests that don't require a real cert.
+        """
+        m = hashlib.sha256()
+        m.update(certificate_pem.encode("utf-8"))
+        return m.hexdigest()
+
+    def compute_der_cert_id(self, certificate_pem: str) -> str:
+        """
+        Compute the certificate hash based on DER encoding
+        """
+        return hashlib.sha256(
+            x509.load_pem_x509_certificate(
+                certificate_pem.encode("utf-8")
+            ).public_bytes(serialization.Encoding.DER)
+        ).hexdigest()
+
     def to_dict(self) -> Dict[str, Any]:
         return {
             "certificateArn": self.arn,
diff --git a/moto/rds/models.py b/moto/rds/models.py
index 5bcc05b61..f880b958f 100644
--- a/moto/rds/models.py
+++ b/moto/rds/models.py
@@ -14,7 +14,7 @@ from moto.core.utils import iso_8601_datetime_with_milliseconds
 from moto.ec2.models import ec2_backends
 from moto.moto_api._internal import mock_random as random
 from moto.neptune.models import NeptuneBackend, neptune_backends
-from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition, load_resource
+from moto.utilities.utils import ARN_PARTITION_REGEX, load_resource
 
 from .exceptions import (
     DBClusterNotFoundError,
@@ -63,26 +63,55 @@ if TYPE_CHECKING:
     from moto.ec2.models.subnets import Subnet
 
 
-def find_cluster(cluster_arn: str) -> "Cluster":
+def find_cluster(cluster_arn: str) -> "DBCluster":
     arn_parts = cluster_arn.split(":")
     region, account = arn_parts[3], arn_parts[4]
     return rds_backends[account][region].describe_db_clusters(cluster_arn)[0]
 
 
-class GlobalCluster(BaseModel):
+class RDSBaseModel(BaseModel):
+    resource_type: str
+
+    def __init__(self, backend: "RDSBackend"):
+        self.backend = backend
+        self.created = iso_8601_datetime_with_milliseconds()
+
+    @property
+    def name(self) -> str:
+        raise NotImplementedError("Subclasses must implement name property.")
+
+    @property
+    def region(self) -> str:
+        return self.backend.region_name
+
+    @property
+    def account_id(self) -> str:
+        return self.backend.account_id
+
+    @property
+    def partition(self) -> str:
+        return self.backend.partition
+
+    @property
+    def arn(self) -> str:
+        return f"arn:{self.partition}:rds:{self.region}:{self.account_id}:{self.resource_type}:{self.name}"
+
+
+class GlobalCluster(RDSBaseModel):
+    resource_type = "global-cluster"
+
     def __init__(
         self,
-        account_id: str,
-        region_name: str,
+        backend: "RDSBackend",
         global_cluster_identifier: str,
         engine: str,
         engine_version: Optional[str],
         storage_encrypted: Optional[str],
         deletion_protection: Optional[str],
     ):
+        super().__init__(backend)
         self.global_cluster_identifier = global_cluster_identifier
         self.global_cluster_resource_id = "cluster-" + random.get_random_hex(8)
-        self.global_cluster_arn = f"arn:{get_partition(region_name)}:rds::{account_id}:global-cluster:{global_cluster_identifier}"
         self.engine = engine
         self.engine_version = engine_version or "5.7.mysql_aurora.2.11.2"
         self.storage_encrypted = (
@@ -95,7 +124,20 @@ class GlobalCluster(BaseModel):
         )
         if self.deletion_protection is None:
             self.deletion_protection = False
-        self.members: List[Cluster] = []
+        self.members: List[DBCluster] = []
+
+    @property
+    def name(self) -> str:
+        return self.global_cluster_identifier
+
+    @property
+    def arn(self) -> str:
+        # Global Clusters do not belong to a particular region.
+        return super().arn.replace(self.region, "")
+
+    @property
+    def global_cluster_arn(self) -> str:
+        return self.arn
 
     def to_xml(self) -> str:
         template = Template(
@@ -111,7 +153,7 @@ class GlobalCluster(BaseModel):
           <GlobalClusterMembers>
           {% for cluster_member in cluster.members %}
           <GlobalClusterMember>
-            <DBClusterArn>{{ cluster_member.db_cluster_arn }}</DBClusterArn>
+            <DBClusterArn>{{ cluster_member.arn }}</DBClusterArn>
             <IsWriter>{{ 'true' if cluster_member.is_writer else 'false' }}</IsWriter>
             {% if not cluster_member.is_writer %}<GlobalWriteForwardingStatus>disabled</GlobalWriteForwardingStatus>{% endif %}
                 <Readers>
@@ -129,7 +171,7 @@ class GlobalCluster(BaseModel):
         return template.render(cluster=self)
 
 
-class Cluster:
+class DBCluster(RDSBaseModel):
     SUPPORTED_FILTERS = {
         "db-cluster-id": FilterDef(
             ["db_cluster_arn", "db_cluster_identifier"], "DB Cluster Identifiers"
@@ -137,9 +179,14 @@ class Cluster:
         "engine": FilterDef(["engine"], "Engine Names"),
     }
 
-    def __init__(self, **kwargs: Any):
+    resource_type = "cluster"
+
+    def __init__(
+        self, backend: "RDSBackend", db_cluster_identifier: str, **kwargs: Any
+    ):
+        super().__init__(backend)
         self.db_name = kwargs.get("db_name")
-        self.db_cluster_identifier = kwargs.get("db_cluster_identifier")
+        self.db_cluster_identifier = db_cluster_identifier
         self.db_cluster_instance_class = kwargs.get("db_cluster_instance_class")
         self.deletion_protection = kwargs.get("deletion_protection")
         if self.deletion_protection is None:
@@ -158,24 +205,22 @@ class Cluster:
             )
         self.engine_version = kwargs.get(
             "engine_version"
-        ) or Cluster.default_engine_version(self.engine)
+        ) or DBCluster.default_engine_version(self.engine)
         self.engine_mode = kwargs.get("engine_mode") or "provisioned"
         self.iops = kwargs.get("iops")
         self.kms_key_id = kwargs.get("kms_key_id")
         self.network_type = kwargs.get("network_type") or "IPV4"
         self.status = "available"
-        self.account_id = kwargs.get("account_id")
-        self.region_name = kwargs["region"]
         self.cluster_create_time = iso_8601_datetime_with_milliseconds()
         self.copy_tags_to_snapshot = kwargs.get("copy_tags_to_snapshot")
         if self.copy_tags_to_snapshot is None:
             self.copy_tags_to_snapshot = False
         self.storage_type = kwargs.get("storage_type")
         if self.storage_type is None:
-            self.storage_type = Cluster.default_storage_type(iops=self.iops)
+            self.storage_type = DBCluster.default_storage_type(iops=self.iops)
         self.allocated_storage = kwargs.get("allocated_storage")
         if self.allocated_storage is None:
-            self.allocated_storage = Cluster.default_allocated_storage(
+            self.allocated_storage = DBCluster.default_allocated_storage(
                 engine=self.engine, storage_type=self.storage_type
             )
         self.master_username = kwargs.get("master_username")
@@ -192,20 +237,20 @@ class Cluster:
         self.availability_zones = kwargs.get("availability_zones")
         if not self.availability_zones:
             self.availability_zones = [
-                f"{self.region_name}a",
-                f"{self.region_name}b",
-                f"{self.region_name}c",
+                f"{self.region}a",
+                f"{self.region}b",
+                f"{self.region}c",
             ]
         self.parameter_group = kwargs.get("parameter_group") or "default.aurora8.0"
         self.subnet_group = kwargs.get("db_subnet_group_name") or "default"
         self.url_identifier = "".join(
             random.choice(string.ascii_lowercase + string.digits) for _ in range(12)
         )
-        self.endpoint = f"{self.db_cluster_identifier}.cluster-{self.url_identifier}.{self.region_name}.rds.amazonaws.com"
-        self.reader_endpoint = f"{self.db_cluster_identifier}.cluster-ro-{self.url_identifier}.{self.region_name}.rds.amazonaws.com"
+        self.endpoint = f"{self.db_cluster_identifier}.cluster-{self.url_identifier}.{self.region}.rds.amazonaws.com"
+        self.reader_endpoint = f"{self.db_cluster_identifier}.cluster-ro-{self.url_identifier}.{self.region}.rds.amazonaws.com"
         self.port: int = kwargs.get("port")  # type: ignore
         if self.port is None:
-            self.port = Cluster.default_port(self.engine)
+            self.port = DBCluster.default_port(self.engine)
         self.preferred_backup_window = (
             kwargs.get("preferred_backup_window") or "01:37-02:07"
         )
@@ -281,6 +326,10 @@ class Cluster:
                     "IAM Authentication is currently not supported by Multi-AZ DB clusters."
                 )
 
+    @property
+    def name(self) -> str:
+        return self.db_cluster_identifier
+
     @property
     def is_multi_az(self) -> bool:
         return (
@@ -288,13 +337,9 @@ class Cluster:
             or self.replication_source_identifier is not None
         )
 
-    @property
-    def arn(self) -> str:
-        return self.db_cluster_arn
-
     @property
     def db_cluster_arn(self) -> str:
-        return f"arn:{get_partition(self.region_name)}:rds:{self.region_name}:{self.account_id}:cluster:{self.db_cluster_identifier}"
+        return self.arn
 
     @property
     def master_user_password(self) -> str:
@@ -349,6 +394,7 @@ class Cluster:
 
     def get_cfg(self) -> Dict[str, Any]:
         cfg = self.__dict__
+        cfg.pop("backend")
         cfg["master_user_password"] = cfg.pop("_master_user_password")
         cfg["enable_http_endpoint"] = cfg.pop("_enable_http_endpoint")
         return cfg
@@ -512,7 +558,9 @@ class Cluster:
         self.tags = [tag_set for tag_set in self.tags if tag_set["Key"] not in tag_keys]
 
 
-class ClusterSnapshot(BaseModel):
+class DBClusterSnapshot(RDSBaseModel):
+    resource_type = "cluster-snapshot"
+
     SUPPORTED_FILTERS = {
         "db-cluster-id": FilterDef(
             ["cluster.db_cluster_arn", "cluster.db_cluster_identifier"],
@@ -527,11 +575,13 @@ class ClusterSnapshot(BaseModel):
 
     def __init__(
         self,
-        cluster: Cluster,
+        backend: "RDSBackend",
+        cluster: DBCluster,
         snapshot_id: str,
         snapshot_type: str,
         tags: List[Dict[str, str]],
     ):
+        super().__init__(backend)
         self.cluster = cluster
         self.snapshot_id = snapshot_id
         self.snapshot_type = snapshot_type
@@ -541,12 +591,12 @@ class ClusterSnapshot(BaseModel):
         self.attributes: List[Dict[str, Any]] = []
 
     @property
-    def arn(self) -> str:
-        return self.snapshot_arn
+    def name(self) -> str:
+        return self.snapshot_id
 
     @property
     def snapshot_arn(self) -> str:
-        return f"arn:{get_partition(self.cluster.region_name)}:rds:{self.cluster.region_name}:{self.cluster.account_id}:cluster-snapshot:{self.snapshot_id}"
+        return self.arn
 
     def to_xml(self) -> str:
         template = Template(
@@ -596,7 +646,7 @@ class ClusterSnapshot(BaseModel):
         self.tags = [tag_set for tag_set in self.tags if tag_set["Key"] not in tag_keys]
 
 
-class Database(CloudFormationModel):
+class DBInstance(CloudFormationModel, RDSBaseModel):
     SUPPORTED_FILTERS = {
         "db-cluster-id": FilterDef(["db_cluster_identifier"], "DB Cluster Identifiers"),
         "db-instance-id": FilterDef(
@@ -620,12 +670,15 @@ class Database(CloudFormationModel):
         "postgres": "9.3.3",
     }
 
-    def __init__(self, **kwargs: Any):
+    resource_type = "db"
+
+    def __init__(
+        self, backend: "RDSBackend", db_instance_identifier: str, **kwargs: Any
+    ):
+        super().__init__(backend)
         self.status = "available"
         self.is_replica = False
         self.replicas: List[str] = []
-        self.account_id: str = kwargs["account_id"]
-        self.region_name: str = kwargs["region"]
         self.engine: str = kwargs["engine"]
         if self.engine not in DbInstanceEngine.valid_db_instance_engine():
             raise InvalidParameterValue(
@@ -642,7 +695,7 @@ class Database(CloudFormationModel):
             self.kms_key_id = kwargs.get("kms_key_id")
         self.storage_type = kwargs.get("storage_type")
         if self.storage_type is None:
-            self.storage_type = Database.default_storage_type(iops=self.iops)
+            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.auto_minor_version_upgrade = kwargs.get("auto_minor_version_upgrade")
@@ -650,17 +703,16 @@ class Database(CloudFormationModel):
             self.auto_minor_version_upgrade = True
         self.allocated_storage = kwargs.get("allocated_storage")
         if self.allocated_storage is None:
-            self.allocated_storage = Database.default_allocated_storage(
+            self.allocated_storage = DBInstance.default_allocated_storage(
                 engine=self.engine, storage_type=self.storage_type
             )
         self.db_cluster_identifier: Optional[str] = kwargs.get("db_cluster_identifier")
-        self.db_instance_identifier = kwargs.get("db_instance_identifier")
+        self.db_instance_identifier = db_instance_identifier
         self.source_db_identifier: Optional[str] = kwargs.get("source_db_identifier")
         self.db_instance_class = kwargs.get("db_instance_class")
         self.port = kwargs.get("port")
         if self.port is None:
-            self.port = Database.default_port(self.engine)
-        self.db_instance_identifier = kwargs.get("db_instance_identifier")
+            self.port = DBInstance.default_port(self.engine)
         self.db_name = kwargs.get("db_name")
         self.instance_create_time = iso_8601_datetime_with_milliseconds()
         self.publicly_accessible = kwargs.get("publicly_accessible")
@@ -674,7 +726,7 @@ class Database(CloudFormationModel):
             self.backup_retention_period = 1
         self.availability_zone = kwargs.get("availability_zone")
         if not self.availability_zone:
-            self.availability_zone = f"{self.region_name}a"
+            self.availability_zone = f"{self.region}a"
         self.multi_az = kwargs.get("multi_az")
         if self.multi_az is None:
             self.multi_az = False
@@ -682,7 +734,7 @@ class Database(CloudFormationModel):
         self.db_subnet_group = None
         if self.db_subnet_group_name:
             self.db_subnet_group = rds_backends[self.account_id][
-                self.region_name
+                self.region
             ].describe_db_subnet_groups(self.db_subnet_group_name)[0]
         self.security_groups = kwargs.get("security_groups", [])
         self.vpc_security_group_ids = kwargs.get("vpc_security_group_ids", [])
@@ -695,22 +747,22 @@ class Database(CloudFormationModel):
         if msg:
             raise RDSClientError("InvalidParameterValue", msg)
 
-        self.db_parameter_group_name = kwargs.get("db_parameter_group_name")
+        self.db_parameter_group_name = kwargs.get("db_parameter_group_name", "")
         if (
             self.db_parameter_group_name
             and not self.is_default_parameter_group(self.db_parameter_group_name)
             and self.db_parameter_group_name
-            not in rds_backends[self.account_id][self.region_name].db_parameter_groups
+            not in rds_backends[self.account_id][self.region].db_parameter_groups
         ):
             raise DBParameterGroupNotFoundError(self.db_parameter_group_name)
 
         self.license_model = kwargs.get("license_model", "general-public-license")
-        self.option_group_name = kwargs.get("option_group_name", None)
+        self.option_group_name = kwargs.get("option_group_name", "")
         self.option_group_supplied = self.option_group_name is not None
         if (
             self.option_group_name
             and self.option_group_name
-            not in rds_backends[self.account_id][self.region_name].option_groups
+            not in rds_backends[self.account_id][self.region].option_groups
         ):
             raise OptionGroupNotFoundFaultError(self.option_group_name)
         self.default_option_groups = {
@@ -736,12 +788,12 @@ class Database(CloudFormationModel):
         )
 
     @property
-    def arn(self) -> str:
-        return self.db_instance_arn
+    def name(self) -> str:
+        return self.db_instance_identifier
 
     @property
     def db_instance_arn(self) -> str:
-        return f"arn:{get_partition(self.region_name)}:rds:{self.region_name}:{self.account_id}:db:{self.db_instance_identifier}"
+        return self.arn
 
     @property
     def physical_resource_id(self) -> Optional[str]:
@@ -758,16 +810,15 @@ class Database(CloudFormationModel):
             description = f"Default parameter group for {db_family}"
             return [
                 DBParameterGroup(
-                    account_id=self.account_id,
+                    backend=self.backend,
                     name=db_parameter_group_name,
                     family=db_family,
                     description=description,
                     tags=[],
-                    region=self.region_name,
                 )
             ]
         else:
-            backend = rds_backends[self.account_id][self.region_name]
+            backend = rds_backends[self.account_id][self.region]
             if self.db_parameter_group_name not in backend.db_parameter_groups:
                 raise DBParameterGroupNotFoundError(self.db_parameter_group_name)
 
@@ -776,9 +827,9 @@ class Database(CloudFormationModel):
     def is_default_parameter_group(self, param_group_name: str) -> bool:
         return param_group_name.startswith(f"default.{self.engine.lower()}")  # type: ignore
 
-    def default_db_parameter_group_details(self) -> Tuple[Optional[str], Optional[str]]:
+    def default_db_parameter_group_details(self) -> Tuple[str, str]:
         if not self.engine_version:
-            return (None, None)
+            return "", ""
 
         minor_engine_version = ".".join(str(self.engine_version).rsplit(".")[:-1])
         db_family = f"{self.engine.lower()}{minor_engine_version}"  # type: ignore
@@ -913,16 +964,18 @@ class Database(CloudFormationModel):
 
     @property
     def address(self) -> str:
-        return f"{self.db_instance_identifier}.aaaaaaaaaa.{self.region_name}.rds.amazonaws.com"
+        return (
+            f"{self.db_instance_identifier}.aaaaaaaaaa.{self.region}.rds.amazonaws.com"
+        )
 
-    def add_replica(self, replica: "Database") -> None:
-        if self.region_name != replica.region_name:
+    def add_replica(self, replica: "DBInstance") -> None:
+        if self.region != replica.region:
             # Cross Region replica
             self.replicas.append(replica.db_instance_arn)
         else:
             self.replicas.append(replica.db_instance_identifier)  # type: ignore
 
-    def remove_replica(self, replica: "Database") -> None:
+    def remove_replica(self, replica: "DBInstance") -> None:
         self.replicas.remove(replica.db_instance_identifier)  # type: ignore
 
     def set_as_replica(self) -> None:
@@ -1010,7 +1063,7 @@ class Database(CloudFormationModel):
         account_id: str,
         region_name: str,
         **kwargs: Any,
-    ) -> "Database":
+    ) -> "DBInstance":
         properties = cloudformation_json["Properties"]
 
         db_security_groups = properties.get("DBSecurityGroups")
@@ -1045,8 +1098,6 @@ class Database(CloudFormationModel):
             "port": properties.get("Port", 3306),
             "publicly_accessible": properties.get("PubliclyAccessible"),
             "copy_tags_to_snapshot": properties.get("CopyTagsToSnapshot"),
-            "account_id": account_id,
-            "region": region_name,
             "security_groups": security_groups,
             "storage_encrypted": properties.get("StorageEncrypted"),
             "storage_type": properties.get("StorageType"),
@@ -1157,7 +1208,8 @@ class Database(CloudFormationModel):
         backend.delete_db_instance(self.db_instance_identifier)  # type: ignore[arg-type]
 
 
-class DatabaseSnapshot(BaseModel):
+class DBInstanceSnapshot(RDSBaseModel):
+    resource_type = "snapshot"
     SUPPORTED_FILTERS = {
         "db-instance-id": FilterDef(
             ["database.db_instance_arn", "database.db_instance_identifier"],
@@ -1171,11 +1223,13 @@ class DatabaseSnapshot(BaseModel):
 
     def __init__(
         self,
-        database: Database,
+        backend: "RDSBackend",
+        database: DBInstance,
         snapshot_id: str,
         snapshot_type: str,
         tags: List[Dict[str, str]],
     ):
+        super().__init__(backend)
         self.database = database
         self.snapshot_id = snapshot_id
         self.snapshot_type = snapshot_type
@@ -1185,12 +1239,12 @@ class DatabaseSnapshot(BaseModel):
         self.attributes: List[Dict[str, Any]] = []
 
     @property
-    def arn(self) -> str:
-        return self.snapshot_arn
+    def name(self) -> str:
+        return self.snapshot_id
 
     @property
     def snapshot_arn(self) -> str:
-        return f"arn:{get_partition(self.database.region_name)}:rds:{self.database.region_name}:{self.database.account_id}:snapshot:{self.snapshot_id}"
+        return self.arn
 
     def to_xml(self) -> str:
         template = Template(
@@ -1258,7 +1312,9 @@ class DatabaseSnapshot(BaseModel):
 
 class ExportTask(BaseModel):
     def __init__(
-        self, snapshot: Union[DatabaseSnapshot, ClusterSnapshot], kwargs: Dict[str, Any]
+        self,
+        snapshot: Union[DBInstanceSnapshot, DBClusterSnapshot],
+        kwargs: Dict[str, Any],
     ):
         self.snapshot = snapshot
 
@@ -1273,7 +1329,7 @@ class ExportTask(BaseModel):
         self.status = "complete"
         self.created_at = iso_8601_datetime_with_milliseconds()
         self.source_type = (
-            "SNAPSHOT" if type(snapshot) is DatabaseSnapshot else "CLUSTER"
+            "SNAPSHOT" if type(snapshot) is DBInstanceSnapshot else "CLUSTER"
         )
 
     def to_xml(self) -> str:
@@ -1306,9 +1362,12 @@ class ExportTask(BaseModel):
         return template.render(task=self, snapshot=self.snapshot)
 
 
-class EventSubscription(BaseModel):
-    def __init__(self, kwargs: Dict[str, Any]):
-        self.subscription_name = kwargs.get("subscription_name")
+class EventSubscription(RDSBaseModel):
+    resource_type = "es"
+
+    def __init__(self, backend: "RDSBackend", subscription_name: str, **kwargs: Any):
+        super().__init__(backend)
+        self.subscription_name = subscription_name
         self.sns_topic_arn = kwargs.get("sns_topic_arn")
         self.source_type = kwargs.get("source_type")
         self.event_categories = kwargs.get("event_categories", [])
@@ -1316,23 +1375,20 @@ class EventSubscription(BaseModel):
         self.enabled = kwargs.get("enabled", False)
         if self.enabled is None:
             self.enabled = False
-        self.tags = kwargs.get("tags", True)
-
-        self.region_name = ""
-        self.customer_aws_id = kwargs["account_id"]
+        self.tags = kwargs.get("tags", [])
         self.status = "active"
         self.created_at = iso_8601_datetime_with_milliseconds()
 
     @property
-    def es_arn(self) -> str:
-        return f"arn:{get_partition(self.region_name)}:rds:{self.region_name}:{self.customer_aws_id}:es:{self.subscription_name}"
+    def name(self) -> str:
+        return self.subscription_name
 
     def to_xml(self) -> str:
         template = Template(
             """
             <EventSubscription>
-              <CustomerAwsId>{{ subscription.customer_aws_id }}</CustomerAwsId>
-              <CustSubscriptionId>{{ subscription.subscription_name }}</CustSubscriptionId>
+              <CustomerAwsId>{{ subscription.account_id }}</CustomerAwsId>
+              <CustSubscriptionId>{{ subscription.name }}</CustSubscriptionId>
               <SnsTopicArn>{{ subscription.sns_topic_arn }}</SnsTopicArn>
               <SubscriptionCreationTime>{{ subscription.created_at }}</SubscriptionCreationTime>
               <SourceType>{{ subscription.source_type }}</SourceType>
@@ -1348,7 +1404,7 @@ class EventSubscription(BaseModel):
               </EventCategoriesList>
               <Status>{{ subscription.status }}</Status>
               <Enabled>{{ 'true' if subscription.enabled else 'false' }}</Enabled>
-              <EventSubscriptionArn>{{ subscription.es_arn }}</EventSubscriptionArn>
+              <EventSubscriptionArn>{{ subscription.arn }}</EventSubscriptionArn>
               <TagList>
               {%- for tag in subscription.tags -%}
                 <Tag><Key>{{ tag['Key'] }}</Key><Value>{{ tag['Value'] }}</Value></Tag>
@@ -1372,23 +1428,29 @@ class EventSubscription(BaseModel):
         self.tags = [tag_set for tag_set in self.tags if tag_set["Key"] not in tag_keys]
 
 
-class SecurityGroup(CloudFormationModel):
+class DBSecurityGroup(CloudFormationModel, RDSBaseModel):
+    resource_type = "secgrp"
+
     def __init__(
         self,
-        account_id: str,
+        backend: "RDSBackend",
         group_name: str,
         description: str,
         tags: List[Dict[str, str]],
     ):
+        super().__init__(backend)
         self.group_name = group_name
         self.description = description
         self.status = "authorized"
         self.ip_ranges: List[Any] = []
         self.ec2_security_groups: List[Any] = []
         self.tags = tags
-        self.owner_id = account_id
         self.vpc_id = None
 
+    @property
+    def name(self) -> str:
+        return self.group_name
+
     def to_xml(self) -> str:
         template = Template(
             """<DBSecurityGroup>
@@ -1396,8 +1458,8 @@ class SecurityGroup(CloudFormationModel):
             {% for security_group in security_group.ec2_security_groups %}
                 <EC2SecurityGroup>
                     <EC2SecurityGroupId>{{ security_group.id }}</EC2SecurityGroupId>
-                    <EC2SecurityGroupName>{{ security_group.name }}</EC2SecurityGroupName>
-                    <EC2SecurityGroupOwnerId>{{ security_group.owner_id }}</EC2SecurityGroupOwnerId>
+                    <EC2SecurityGroupName>{{ security_group.group_name }}</EC2SecurityGroupName>
+                    <EC2SecurityGroupOwnerId>{{ security_group.account_id }}</EC2SecurityGroupOwnerId>
                     <Status>authorized</Status>
                 </EC2SecurityGroup>
             {% endfor %}
@@ -1413,7 +1475,8 @@ class SecurityGroup(CloudFormationModel):
             {% endfor %}
             </IPRanges>
             <OwnerId>{{ security_group.ownder_id }}</OwnerId>
-            <DBSecurityGroupName>{{ security_group.group_name }}</DBSecurityGroupName>
+            <DBSecurityGroupName>{{ security_group.name }}</DBSecurityGroupName>
+            <DBSecurityGroupArn>{{ security_group.arn }}</DBSecurityGroupArn>
         </DBSecurityGroup>"""
         )
         return template.render(security_group=self)
@@ -1458,7 +1521,7 @@ class SecurityGroup(CloudFormationModel):
         account_id: str,
         region_name: str,
         **kwargs: Any,
-    ) -> "SecurityGroup":
+    ) -> "DBSecurityGroup":
         properties = cloudformation_json["Properties"]
         group_name = resource_name.lower()
         description = properties["GroupDescription"]
@@ -1499,28 +1562,28 @@ class SecurityGroup(CloudFormationModel):
         backend.delete_security_group(self.group_name)
 
 
-class SubnetGroup(CloudFormationModel):
+class DBSubnetGroup(CloudFormationModel, RDSBaseModel):
+    resource_type = "subgrp"
+
     def __init__(
         self,
+        backend: "RDSBackend",
         subnet_name: str,
         description: str,
         subnets: "List[Subnet]",
         tags: List[Dict[str, str]],
-        region: str,
-        account_id: str,
     ):
+        super().__init__(backend)
         self.subnet_name = subnet_name
         self.description = description
         self.subnets = subnets
         self.status = "Complete"
         self.tags = tags
         self.vpc_id = self.subnets[0].vpc_id
-        self.region = region
-        self.account_id = account_id
 
     @property
-    def sg_arn(self) -> str:
-        return f"arn:{get_partition(self.region)}:rds:{self.region}:{self.account_id}:subgrp:{self.subnet_name}"
+    def name(self) -> str:
+        return self.subnet_name
 
     def to_xml(self) -> str:
         template = Template(
@@ -1528,8 +1591,8 @@ class SubnetGroup(CloudFormationModel):
               <VpcId>{{ subnet_group.vpc_id }}</VpcId>
               <SubnetGroupStatus>{{ subnet_group.status }}</SubnetGroupStatus>
               <DBSubnetGroupDescription>{{ subnet_group.description }}</DBSubnetGroupDescription>
-              <DBSubnetGroupName>{{ subnet_group.subnet_name }}</DBSubnetGroupName>
-              <DBSubnetGroupArn>{{ subnet_group.sg_arn }}</DBSubnetGroupArn>
+              <DBSubnetGroupName>{{ subnet_group.name }}</DBSubnetGroupName>
+              <DBSubnetGroupArn>{{ subnet_group.arn }}</DBSubnetGroupArn>
               <Subnets>
                 {% for subnet in subnet_group.subnets %}
                 <Subnet>
@@ -1586,7 +1649,7 @@ class SubnetGroup(CloudFormationModel):
         account_id: str,
         region_name: str,
         **kwargs: Any,
-    ) -> "SubnetGroup":
+    ) -> "DBSubnetGroup":
         properties = cloudformation_json["Properties"]
 
         description = properties["DBSubnetGroupDescription"]
@@ -1621,22 +1684,24 @@ class SubnetGroup(CloudFormationModel):
         backend.delete_subnet_group(self.subnet_name)
 
 
-class DBProxy(BaseModel):
+class DBProxy(RDSBaseModel):
+    resource_type = "db-proxy"
+
     def __init__(
         self,
+        backend: "RDSBackend",
         db_proxy_name: str,
         engine_family: str,
         auth: List[Dict[str, str]],
         role_arn: str,
         vpc_subnet_ids: List[str],
-        region_name: str,
-        account_id: str,
         vpc_security_group_ids: Optional[List[str]],
         require_tls: Optional[bool] = False,
         idle_client_timeout: Optional[int] = 1800,
         debug_logging: Optional[bool] = False,
         tags: Optional[List[Dict[str, str]]] = None,
     ):
+        super().__init__(backend)
         self.db_proxy_name = db_proxy_name
         self.engine_family = engine_family
         if self.engine_family not in ["MYSQL", "POSTGRESQ", "SQLSERVER"]:
@@ -1664,11 +1729,7 @@ class DBProxy(BaseModel):
             self.tags = []
         else:
             self.tags = tags
-        self.region_name = region_name
-        self.account_id = account_id
-        self.db_proxy_arn = f"arn:{get_partition(self.region_name)}:rds:{self.region_name}:{self.account_id}:db-proxy:{self.db_proxy_name}"
-        self.arn = self.db_proxy_arn
-        ec2_backend = ec2_backends[self.account_id][self.region_name]
+        ec2_backend = ec2_backends[self.account_id][self.region]
         subnets = ec2_backend.describe_subnets(subnet_ids=self.vpc_subnet_ids)
         vpcs = []
         for subnet in subnets:
@@ -1679,11 +1740,15 @@ class DBProxy(BaseModel):
         self.vpc_id = ec2_backend.describe_subnets(subnet_ids=[self.vpc_subnet_ids[0]])[
             0
         ].vpc_id
-        self.status = "availible"
+        self.status = "available"
         self.url_identifier = "".join(
             random.choice(string.ascii_lowercase + string.digits) for _ in range(12)
         )
-        self.endpoint = f"{self.db_proxy_name}.db-proxy-{self.url_identifier}.{self.region_name}.rds.amazonaws.com"
+        self.endpoint = f"{self.db_proxy_name}.db-proxy-{self.url_identifier}.{self.region}.rds.amazonaws.com"
+
+    @property
+    def name(self) -> str:
+        return self.db_proxy_name
 
     def get_tags(self) -> List[Dict[str, str]]:
         return self.tags
@@ -1729,7 +1794,7 @@ class DBProxy(BaseModel):
                 <RoleArn>{{ dbproxy.role_arn }}</RoleArn>
                 <DebugLogging>{{ 'true' if dbproxy.debug_logging else 'false' }}</DebugLogging>
                 <VpcId>{{ dbproxy.vpc_id }}</VpcId>
-                <DBProxyArn>{{ dbproxy.db_proxy_arn }}</DBProxyArn>
+                <DBProxyArn>{{ dbproxy.arn }}</DBProxyArn>
                 <VpcSubnetIds>
                   {% for vpcsubnetid in dbproxy.vpc_subnet_ids %}
                     <member>{{ vpcsubnetid }}</member>
@@ -1748,18 +1813,18 @@ class RDSBackend(BaseBackend):
             ARN_PARTITION_REGEX
             + r":rds:.*:[0-9]*:(db|cluster|es|og|pg|ri|secgrp|snapshot|cluster-snapshot|subgrp|db-proxy):.*$"
         )
-        self.clusters: Dict[str, Cluster] = OrderedDict()
+        self.clusters: Dict[str, DBCluster] = OrderedDict()
         self.global_clusters: Dict[str, GlobalCluster] = OrderedDict()
-        self.databases: Dict[str, Database] = OrderedDict()
-        self.database_snapshots: Dict[str, DatabaseSnapshot] = OrderedDict()
-        self.cluster_snapshots: Dict[str, ClusterSnapshot] = OrderedDict()
+        self.databases: Dict[str, DBInstance] = OrderedDict()
+        self.database_snapshots: Dict[str, DBInstanceSnapshot] = OrderedDict()
+        self.cluster_snapshots: Dict[str, DBClusterSnapshot] = OrderedDict()
         self.export_tasks: Dict[str, ExportTask] = OrderedDict()
         self.event_subscriptions: Dict[str, EventSubscription] = OrderedDict()
         self.db_parameter_groups: Dict[str, DBParameterGroup] = {}
         self.db_cluster_parameter_groups: Dict[str, DBClusterParameterGroup] = {}
         self.option_groups: Dict[str, OptionGroup] = {}
-        self.security_groups: Dict[str, SecurityGroup] = {}
-        self.subnet_groups: Dict[str, SubnetGroup] = {}
+        self.security_groups: Dict[str, DBSecurityGroup] = {}
+        self.subnet_groups: Dict[str, DBSubnetGroup] = {}
         self._db_cluster_options: Optional[List[Dict[str, Any]]] = None
         self.db_proxies: Dict[str, DBProxy] = OrderedDict()
 
@@ -1784,10 +1849,10 @@ class RDSBackend(BaseBackend):
             ]
         return self._db_cluster_options
 
-    def create_db_instance(self, db_kwargs: Dict[str, Any]) -> Database:
+    def create_db_instance(self, db_kwargs: Dict[str, Any]) -> DBInstance:
         database_id = db_kwargs["db_instance_identifier"]
         self._validate_db_identifier(database_id)
-        database = Database(**db_kwargs)
+        database = DBInstance(self, **db_kwargs)
 
         cluster_id = database.db_cluster_identifier
         if cluster_id is not None:
@@ -1812,18 +1877,18 @@ class RDSBackend(BaseBackend):
         self,
         db_instance_identifier: str,
         db_snapshot_identifier: str,
-    ) -> DatabaseSnapshot:
+    ) -> DBInstanceSnapshot:
         return self.create_db_snapshot(
             db_instance_identifier, db_snapshot_identifier, snapshot_type="automated"
         )
 
     def create_db_snapshot(
         self,
-        db_instance: Union[str, Database],
+        db_instance: Union[str, DBInstance],
         db_snapshot_identifier: str,
         snapshot_type: str = "manual",
         tags: Optional[List[Dict[str, str]]] = None,
-    ) -> DatabaseSnapshot:
+    ) -> DBInstanceSnapshot:
         if isinstance(db_instance, str):
             database = self.databases.get(db_instance)
             if not database:
@@ -1841,8 +1906,8 @@ class RDSBackend(BaseBackend):
             tags = list()
         if database.copy_tags_to_snapshot and not tags:
             tags = database.get_tags()
-        snapshot = DatabaseSnapshot(
-            database, db_snapshot_identifier, snapshot_type, tags
+        snapshot = DBInstanceSnapshot(
+            self, database, db_snapshot_identifier, snapshot_type, tags
         )
         self.database_snapshots[db_snapshot_identifier] = snapshot
         return snapshot
@@ -1852,7 +1917,7 @@ class RDSBackend(BaseBackend):
         source_snapshot_identifier: str,
         target_snapshot_identifier: str,
         tags: Optional[List[Dict[str, str]]] = None,
-    ) -> DatabaseSnapshot:
+    ) -> DBInstanceSnapshot:
         if source_snapshot_identifier not in self.database_snapshots:
             raise DBSnapshotNotFoundError(source_snapshot_identifier)
 
@@ -1867,13 +1932,13 @@ class RDSBackend(BaseBackend):
             tags=tags,
         )
 
-    def delete_db_snapshot(self, db_snapshot_identifier: str) -> DatabaseSnapshot:
+    def delete_db_snapshot(self, db_snapshot_identifier: str) -> DBInstanceSnapshot:
         if db_snapshot_identifier not in self.database_snapshots:
             raise DBSnapshotNotFoundError(db_snapshot_identifier)
 
         return self.database_snapshots.pop(db_snapshot_identifier)
 
-    def promote_read_replica(self, db_kwargs: Dict[str, Any]) -> Database:
+    def promote_read_replica(self, db_kwargs: Dict[str, Any]) -> DBInstance:
         database_id = db_kwargs["db_instance_identifier"]
         database = self.databases[database_id]
         if database.is_replica:
@@ -1882,17 +1947,16 @@ class RDSBackend(BaseBackend):
 
         return database
 
-    def create_db_instance_read_replica(self, db_kwargs: Dict[str, Any]) -> Database:
+    def create_db_instance_read_replica(self, db_kwargs: Dict[str, Any]) -> DBInstance:
         database_id = db_kwargs["db_instance_identifier"]
         source_database_id = db_kwargs["source_db_identifier"]
         primary = self.find_db_from_id(source_database_id)
         if self.arn_regex.match(source_database_id):
-            db_kwargs["region"] = self.region_name
+            db_kwargs["backend"] = self
 
         # Shouldn't really copy here as the instance is duplicated. RDS replicas have different instances.
         replica = copy.copy(primary)
         replica.update(db_kwargs)
-        replica.region_name = self.region_name
         replica.set_as_replica()
         self.databases[database_id] = replica
         primary.add_replica(replica)
@@ -1900,14 +1964,14 @@ class RDSBackend(BaseBackend):
 
     def describe_db_instances(
         self, db_instance_identifier: Optional[str] = None, filters: Any = None
-    ) -> List[Database]:
+    ) -> List[DBInstance]:
         databases = self.databases
         if db_instance_identifier:
             filters = merge_filters(
                 filters, {"db-instance-id": [db_instance_identifier]}
             )
         if filters:
-            databases = self._filter_resources(databases, filters, Database)
+            databases = self._filter_resources(databases, filters, DBInstance)
         if db_instance_identifier and not databases:
             raise DBInstanceNotFoundError(db_instance_identifier)
         return list(databases.values())
@@ -1917,7 +1981,7 @@ class RDSBackend(BaseBackend):
         db_instance_identifier: Optional[str],
         db_snapshot_identifier: str,
         filters: Optional[Dict[str, Any]] = None,
-    ) -> List[DatabaseSnapshot]:
+    ) -> List[DBInstanceSnapshot]:
         snapshots = self.database_snapshots
         if db_instance_identifier:
             filters = merge_filters(
@@ -1928,14 +1992,14 @@ class RDSBackend(BaseBackend):
                 filters, {"db-snapshot-id": [db_snapshot_identifier]}
             )
         if filters:
-            snapshots = self._filter_resources(snapshots, filters, DatabaseSnapshot)
+            snapshots = self._filter_resources(snapshots, filters, DBInstanceSnapshot)
         if db_snapshot_identifier and not snapshots and not db_instance_identifier:
             raise DBSnapshotNotFoundError(db_snapshot_identifier)
         return list(snapshots.values())
 
     def modify_db_instance(
         self, db_instance_identifier: str, db_kwargs: Dict[str, Any]
-    ) -> Database:
+    ) -> DBInstance:
         database = self.describe_db_instances(db_instance_identifier)[0]
         if "new_db_instance_identifier" in db_kwargs:
             del self.databases[db_instance_identifier]
@@ -1953,17 +2017,18 @@ class RDSBackend(BaseBackend):
         database.update(db_kwargs)
         return database
 
-    def reboot_db_instance(self, db_instance_identifier: str) -> Database:
+    def reboot_db_instance(self, db_instance_identifier: str) -> DBInstance:
         return self.describe_db_instances(db_instance_identifier)[0]
 
     def restore_db_instance_from_db_snapshot(
         self, from_snapshot_id: str, overrides: Dict[str, Any]
-    ) -> Database:
+    ) -> DBInstance:
         snapshot = self.describe_db_snapshots(
             db_instance_identifier=None, db_snapshot_identifier=from_snapshot_id
         )[0]
         original_database = snapshot.database
         new_instance_props = copy.deepcopy(original_database.__dict__)
+        new_instance_props.pop("backend")
         if not original_database.option_group_supplied:
             # If the option group is not supplied originally, the 'option_group_name' will receive a default value
             # Force this reconstruction, and prevent any validation on the default value
@@ -1980,7 +2045,7 @@ class RDSBackend(BaseBackend):
         source_db_identifier: str,
         target_db_identifier: str,
         overrides: Dict[str, Any],
-    ) -> Database:
+    ) -> DBInstance:
         db_instance = self.describe_db_instances(
             db_instance_identifier=source_db_identifier
         )[0]
@@ -1991,6 +2056,7 @@ class RDSBackend(BaseBackend):
         del source_dict["db_subnet_group"]
 
         new_instance_props = copy.deepcopy(source_dict)
+        new_instance_props.pop("backend")
         if not db_instance.option_group_supplied:
             # If the option group is not supplied originally, the 'option_group_name' will receive a default value
             # Force this reconstruction, and prevent any validation on the default value
@@ -2007,7 +2073,7 @@ class RDSBackend(BaseBackend):
 
     def stop_db_instance(
         self, db_instance_identifier: str, db_snapshot_identifier: Optional[str] = None
-    ) -> Database:
+    ) -> DBInstance:
         self._validate_db_identifier(db_instance_identifier)
         database = self.describe_db_instances(db_instance_identifier)[0]
         # todo: certain rds types not allowed to be stopped at this time.
@@ -2024,7 +2090,7 @@ class RDSBackend(BaseBackend):
         database.status = "stopped"
         return database
 
-    def start_db_instance(self, db_instance_identifier: str) -> Database:
+    def start_db_instance(self, db_instance_identifier: str) -> DBInstance:
         self._validate_db_identifier(db_instance_identifier)
         database = self.describe_db_instances(db_instance_identifier)[0]
         # todo: bunch of different error messages to be generated from this api call
@@ -2033,7 +2099,7 @@ class RDSBackend(BaseBackend):
         database.status = "available"
         return database
 
-    def find_db_from_id(self, db_id: str) -> Database:
+    def find_db_from_id(self, db_id: str) -> DBInstance:
         if self.arn_regex.match(db_id):
             arn_breakdown = db_id.split(":")
             region = arn_breakdown[3]
@@ -2047,7 +2113,7 @@ class RDSBackend(BaseBackend):
 
     def delete_db_instance(
         self, db_instance_identifier: str, db_snapshot_name: Optional[str] = None
-    ) -> Database:
+    ) -> DBInstance:
         self._validate_db_identifier(db_instance_identifier)
         if db_instance_identifier in self.databases:
             if self.databases[db_instance_identifier].deletion_protection:
@@ -2071,12 +2137,14 @@ class RDSBackend(BaseBackend):
 
     def create_db_security_group(
         self, group_name: str, description: str, tags: List[Dict[str, str]]
-    ) -> SecurityGroup:
-        security_group = SecurityGroup(self.account_id, group_name, description, tags)
+    ) -> DBSecurityGroup:
+        security_group = DBSecurityGroup(self, group_name, description, tags)
         self.security_groups[group_name] = security_group
         return security_group
 
-    def describe_security_groups(self, security_group_name: str) -> List[SecurityGroup]:
+    def describe_security_groups(
+        self, security_group_name: str
+    ) -> List[DBSecurityGroup]:
         if security_group_name:
             if security_group_name in self.security_groups:
                 return [self.security_groups[security_group_name]]
@@ -2084,7 +2152,7 @@ class RDSBackend(BaseBackend):
                 raise DBSecurityGroupNotFoundError(security_group_name)
         return list(self.security_groups.values())
 
-    def delete_security_group(self, security_group_name: str) -> SecurityGroup:
+    def delete_security_group(self, security_group_name: str) -> DBSecurityGroup:
         if security_group_name in self.security_groups:
             return self.security_groups.pop(security_group_name)
         else:
@@ -2100,7 +2168,7 @@ class RDSBackend(BaseBackend):
 
     def authorize_security_group(
         self, security_group_name: str, cidr_ip: str
-    ) -> SecurityGroup:
+    ) -> DBSecurityGroup:
         security_group = self.describe_security_groups(security_group_name)[0]
         security_group.authorize_cidr(cidr_ip)
         return security_group
@@ -2111,14 +2179,12 @@ class RDSBackend(BaseBackend):
         description: str,
         subnets: List[Any],
         tags: List[Dict[str, str]],
-    ) -> SubnetGroup:
-        subnet_group = SubnetGroup(
-            subnet_name, description, subnets, tags, self.region_name, self.account_id
-        )
+    ) -> DBSubnetGroup:
+        subnet_group = DBSubnetGroup(self, subnet_name, description, subnets, tags)
         self.subnet_groups[subnet_name] = subnet_group
         return subnet_group
 
-    def describe_db_subnet_groups(self, subnet_group_name: str) -> List[SubnetGroup]:
+    def describe_db_subnet_groups(self, subnet_group_name: str) -> List[DBSubnetGroup]:
         if subnet_group_name:
             if subnet_group_name in self.subnet_groups:
                 return [self.subnet_groups[subnet_group_name]]
@@ -2128,7 +2194,7 @@ class RDSBackend(BaseBackend):
 
     def modify_db_subnet_group(
         self, subnet_name: str, description: str, subnets: "List[Subnet]"
-    ) -> SubnetGroup:
+    ) -> DBSubnetGroup:
         subnet_group = self.subnet_groups.pop(subnet_name)
         if not subnet_group:
             raise DBSubnetGroupNotFoundError(subnet_name)
@@ -2138,7 +2204,7 @@ class RDSBackend(BaseBackend):
             subnet_group.description = description
         return subnet_group
 
-    def delete_subnet_group(self, subnet_name: str) -> SubnetGroup:
+    def delete_subnet_group(self, subnet_name: str) -> DBSubnetGroup:
         if subnet_name in self.subnet_groups:
             return self.subnet_groups.pop(subnet_name)
         else:
@@ -2208,10 +2274,7 @@ class RDSBackend(BaseBackend):
         # }
         # The CLI does not allow deletion of default groups
 
-        # add region and account-id to construct the arn
-        option_group_kwargs["region"] = self.region_name
-        option_group_kwargs["account_id"] = self.account_id
-        option_group = OptionGroup(**option_group_kwargs)
+        option_group = OptionGroup(self, **option_group_kwargs)
         self.option_groups[option_group_id] = option_group
         return option_group
 
@@ -2348,9 +2411,7 @@ class RDSBackend(BaseBackend):
                 "InvalidParameterValue",
                 "The parameter DBParameterGroupName must be provided and must not be blank.",
             )
-        db_parameter_group_kwargs["region"] = self.region_name
-        db_parameter_group_kwargs["account_id"] = self.account_id
-        db_parameter_group = DBParameterGroup(**db_parameter_group_kwargs)
+        db_parameter_group = DBParameterGroup(self, **db_parameter_group_kwargs)
         self.db_parameter_groups[db_parameter_group_id] = db_parameter_group
         return db_parameter_group
 
@@ -2402,10 +2463,9 @@ class RDSBackend(BaseBackend):
     def describe_db_cluster_parameters(self) -> List[Dict[str, Any]]:
         return []
 
-    def create_db_cluster(self, kwargs: Dict[str, Any]) -> Cluster:
+    def create_db_cluster(self, kwargs: Dict[str, Any]) -> DBCluster:
         cluster_id = kwargs["db_cluster_identifier"]
-        kwargs["account_id"] = self.account_id
-        cluster = Cluster(**kwargs)
+        cluster = DBCluster(self, **kwargs)
         self.clusters[cluster_id] = cluster
 
         if (
@@ -2444,7 +2504,7 @@ class RDSBackend(BaseBackend):
         cluster.status = "available"  # Already set the final status in the background
         return initial_state
 
-    def modify_db_cluster(self, kwargs: Dict[str, Any]) -> Cluster:
+    def modify_db_cluster(self, kwargs: Dict[str, Any]) -> DBCluster:
         cluster_id = kwargs["db_cluster_identifier"]
 
         if cluster_id in self.neptune.clusters:
@@ -2472,7 +2532,7 @@ class RDSBackend(BaseBackend):
         cluster.status = "available"  # Already set the final status in the background
         return initial_state
 
-    def promote_read_replica_db_cluster(self, db_cluster_identifier: str) -> Cluster:
+    def promote_read_replica_db_cluster(self, db_cluster_identifier: str) -> DBCluster:
         cluster = self.clusters[db_cluster_identifier]
         source_cluster = find_cluster(cluster.replication_source_identifier)  # type: ignore
         source_cluster.read_replica_identifiers.remove(cluster.db_cluster_arn)
@@ -2481,7 +2541,7 @@ class RDSBackend(BaseBackend):
 
     def create_auto_cluster_snapshot(
         self, db_cluster_identifier: str, db_snapshot_identifier: str
-    ) -> ClusterSnapshot:
+    ) -> DBClusterSnapshot:
         return self.create_db_cluster_snapshot(
             db_cluster_identifier, db_snapshot_identifier, snapshot_type="automated"
         )
@@ -2492,7 +2552,7 @@ class RDSBackend(BaseBackend):
         db_snapshot_identifier: str,
         snapshot_type: str = "manual",
         tags: Optional[List[Dict[str, str]]] = None,
-    ) -> ClusterSnapshot:
+    ) -> DBClusterSnapshot:
         cluster = self.clusters.get(db_cluster_identifier)
         if cluster is None:
             raise DBClusterNotFoundError(db_cluster_identifier)
@@ -2506,7 +2566,9 @@ class RDSBackend(BaseBackend):
             tags = list()
         if cluster.copy_tags_to_snapshot:
             tags += cluster.get_tags()
-        snapshot = ClusterSnapshot(cluster, db_snapshot_identifier, snapshot_type, tags)
+        snapshot = DBClusterSnapshot(
+            self, cluster, db_snapshot_identifier, snapshot_type, tags
+        )
         self.cluster_snapshots[db_snapshot_identifier] = snapshot
         return snapshot
 
@@ -2515,7 +2577,7 @@ class RDSBackend(BaseBackend):
         source_snapshot_identifier: str,
         target_snapshot_identifier: str,
         tags: Optional[List[Dict[str, str]]] = None,
-    ) -> ClusterSnapshot:
+    ) -> DBClusterSnapshot:
         if source_snapshot_identifier not in self.cluster_snapshots:
             raise DBClusterSnapshotNotFoundError(source_snapshot_identifier)
 
@@ -2532,7 +2594,7 @@ class RDSBackend(BaseBackend):
 
     def delete_db_cluster_snapshot(
         self, db_snapshot_identifier: str
-    ) -> ClusterSnapshot:
+    ) -> DBClusterSnapshot:
         if db_snapshot_identifier not in self.cluster_snapshots:
             raise DBClusterSnapshotNotFoundError(db_snapshot_identifier)
 
@@ -2540,15 +2602,15 @@ class RDSBackend(BaseBackend):
 
     def describe_db_clusters(
         self, cluster_identifier: Optional[str] = None, filters: Any = None
-    ) -> List[Cluster]:
+    ) -> List[DBCluster]:
         clusters = self.clusters
         clusters_neptune = self.neptune.clusters
         if cluster_identifier:
             filters = merge_filters(filters, {"db-cluster-id": [cluster_identifier]})
         if filters:
-            clusters = self._filter_resources(clusters, filters, Cluster)
+            clusters = self._filter_resources(clusters, filters, DBCluster)
             clusters_neptune = self._filter_resources(
-                clusters_neptune, filters, Cluster
+                clusters_neptune, filters, DBCluster
             )
         if cluster_identifier and not (clusters or clusters_neptune):
             raise DBClusterNotFoundError(cluster_identifier)
@@ -2559,7 +2621,7 @@ class RDSBackend(BaseBackend):
         db_cluster_identifier: Optional[str],
         db_snapshot_identifier: str,
         filters: Any = None,
-    ) -> List[ClusterSnapshot]:
+    ) -> List[DBClusterSnapshot]:
         snapshots = self.cluster_snapshots
         if db_cluster_identifier:
             filters = merge_filters(filters, {"db-cluster-id": [db_cluster_identifier]})
@@ -2568,14 +2630,14 @@ class RDSBackend(BaseBackend):
                 filters, {"db-cluster-snapshot-id": [db_snapshot_identifier]}
             )
         if filters:
-            snapshots = self._filter_resources(snapshots, filters, ClusterSnapshot)
+            snapshots = self._filter_resources(snapshots, filters, DBClusterSnapshot)
         if db_snapshot_identifier and not snapshots and not db_cluster_identifier:
             raise DBClusterSnapshotNotFoundError(db_snapshot_identifier)
         return list(snapshots.values())
 
     def delete_db_cluster(
         self, cluster_identifier: str, snapshot_name: Optional[str] = None
-    ) -> Cluster:
+    ) -> DBCluster:
         if cluster_identifier in self.clusters:
             cluster = self.clusters[cluster_identifier]
             if cluster.deletion_protection:
@@ -2595,7 +2657,7 @@ class RDSBackend(BaseBackend):
             return self.neptune.delete_db_cluster(cluster_identifier)  # type: ignore
         raise DBClusterNotFoundError(cluster_identifier)
 
-    def start_db_cluster(self, cluster_identifier: str) -> Cluster:
+    def start_db_cluster(self, cluster_identifier: str) -> DBCluster:
         if cluster_identifier not in self.clusters:
             return self.neptune.start_db_cluster(cluster_identifier)  # type: ignore
             raise DBClusterNotFoundError(cluster_identifier)
@@ -2611,7 +2673,7 @@ class RDSBackend(BaseBackend):
 
     def restore_db_cluster_from_snapshot(
         self, from_snapshot_id: str, overrides: Dict[str, Any]
-    ) -> Cluster:
+    ) -> DBCluster:
         snapshot = self.describe_db_cluster_snapshots(
             db_cluster_identifier=None, db_snapshot_identifier=from_snapshot_id
         )[0]
@@ -2623,7 +2685,7 @@ class RDSBackend(BaseBackend):
 
         return self.create_db_cluster(new_cluster_props)
 
-    def stop_db_cluster(self, cluster_identifier: str) -> "Cluster":
+    def stop_db_cluster(self, cluster_identifier: str) -> "DBCluster":
         if cluster_identifier not in self.clusters:
             raise DBClusterNotFoundError(cluster_identifier)
         cluster = self.clusters[cluster_identifier]
@@ -2652,7 +2714,7 @@ class RDSBackend(BaseBackend):
             raise DBClusterSnapshotNotFoundError(snapshot_id)
 
         if snapshot_type == "snapshot":
-            snapshot: Union[DatabaseSnapshot, ClusterSnapshot] = (
+            snapshot: Union[DBInstanceSnapshot, DBClusterSnapshot] = (
                 self.database_snapshots[snapshot_id]
             )
         else:
@@ -2690,8 +2752,7 @@ class RDSBackend(BaseBackend):
         if subscription_name in self.event_subscriptions:
             raise SubscriptionAlreadyExistError(subscription_name)
 
-        kwargs["account_id"] = self.account_id
-        subscription = EventSubscription(kwargs)
+        subscription = EventSubscription(self, **kwargs)
         self.event_subscriptions[subscription_name] = subscription
 
         return subscription
@@ -2907,8 +2968,7 @@ class RDSBackend(BaseBackend):
         description: str,
     ) -> "DBClusterParameterGroup":
         group = DBClusterParameterGroup(
-            account_id=self.account_id,
-            region=self.region_name,
+            backend=self,
             name=group_name,
             family=family,
             description=description,
@@ -2957,8 +3017,7 @@ class RDSBackend(BaseBackend):
                 "When creating standalone global cluster, value for engineName should be specified"
             )
         global_cluster = GlobalCluster(
-            account_id=self.account_id,
-            region_name=self.region_name,
+            backend=self,
             global_cluster_identifier=global_cluster_identifier,
             engine=engine,  # type: ignore
             engine_version=engine_version,
@@ -3106,13 +3165,12 @@ class RDSBackend(BaseBackend):
         if len(self.db_proxies) >= int(os.environ.get("MOTO_RDS_PROXY_LIMIT", "100")):
             raise DBProxyQuotaExceededFault()
         db_proxy = DBProxy(
+            self,
             db_proxy_name,
             engine_family,
             auth,
             role_arn,
             vpc_subnet_ids,
-            self.region_name,
-            self.account_id,
             vpc_security_group_ids,
             require_tls,
             idle_client_timeout,
@@ -3138,25 +3196,30 @@ class RDSBackend(BaseBackend):
         return db_proxies
 
 
-class OptionGroup:
+class OptionGroup(RDSBaseModel):
+    resource_type = "og"
+
     def __init__(
         self,
+        backend: "RDSBackend",
         name: str,
         engine_name: str,
         major_engine_version: str,
-        region: str,
-        account_id: str,
         description: Optional[str] = None,
     ):
+        super().__init__(backend)
         self.engine_name = engine_name
         self.major_engine_version = major_engine_version
         self.description = description
-        self.name = name
+        self._name = name
         self.vpc_and_non_vpc_instance_memberships = False
         self.options: Dict[str, Any] = {}
         self.vpcId = "null"
         self.tags: List[Dict[str, str]] = []
-        self.arn = f"arn:{get_partition(region)}:rds:{region}:{account_id}:og:{name}"
+
+    @property
+    def name(self) -> str:
+        return self._name
 
     def to_json(self) -> str:
         template = Template(
@@ -3216,22 +3279,27 @@ class OptionGroup:
         self.tags = [tag_set for tag_set in self.tags if tag_set["Key"] not in tag_keys]
 
 
-class DBParameterGroup(CloudFormationModel):
+class DBParameterGroup(CloudFormationModel, RDSBaseModel):
+    resource_type = "pg"
+
     def __init__(
         self,
-        account_id: str,
-        name: Optional[str],
+        backend: "RDSBackend",
+        name: str,
         description: str,
         family: Optional[str],
         tags: List[Dict[str, str]],
-        region: str,
     ):
-        self.name = name
+        super().__init__(backend)
+        self._name = name
         self.description = description
         self.family = family
         self.tags = tags
         self.parameters: Dict[str, Any] = defaultdict(dict)
-        self.arn = f"arn:{get_partition(region)}:rds:{region}:{account_id}:pg:{name}"
+
+    @property
+    def name(self) -> str:
+        return self._name
 
     def to_xml(self) -> str:
         template = Template(
@@ -3307,15 +3375,19 @@ class DBParameterGroup(CloudFormationModel):
         return db_parameter_group
 
 
-class DBClusterParameterGroup(CloudFormationModel):
-    def __init__(
-        self, account_id: str, region: str, name: str, description: str, family: str
-    ):
-        self.name = name
+class DBClusterParameterGroup(CloudFormationModel, RDSBaseModel):
+    resource_type = "cluster-pg"
+
+    def __init__(self, backend: RDSBackend, name: str, description: str, family: str):
+        super().__init__(backend)
+        self._name = name
         self.description = description
         self.family = family
         self.parameters: Dict[str, Any] = defaultdict(dict)
-        self.arn = f"arn:{get_partition(region)}:rds:{region}:{account_id}:cpg:{name}"
+
+    @property
+    def name(self) -> str:
+        return self._name
 
     def to_xml(self) -> str:
         template = Template(
diff --git a/moto/rds/responses.py b/moto/rds/responses.py
index 6c55a5989..635851c89 100644
--- a/moto/rds/responses.py
+++ b/moto/rds/responses.py
@@ -75,8 +75,6 @@ class RDSResponse(BaseResponse):
                 "PreferredMaintenanceWindow", "wed:06:38-wed:07:08"
             ).lower(),
             "publicly_accessible": self._get_param("PubliclyAccessible"),
-            "account_id": self.current_account,
-            "region": self.region,
             "security_groups": self._get_multi_param(
                 "DBSecurityGroups.DBSecurityGroupName"
             ),
@@ -132,8 +130,6 @@ class RDSResponse(BaseResponse):
                 "PreferredMaintenanceWindow"
             ),
             "publicly_accessible": self._get_param("PubliclyAccessible"),
-            "account_id": self.current_account,
-            "region": self.region,
             "security_groups": self._get_multi_param(
                 "DBSecurityGroups.DBSecurityGroupName"
             ),
@@ -211,7 +207,6 @@ class RDSResponse(BaseResponse):
             "network_type": self._get_param("NetworkType"),
             "port": self._get_param("Port"),
             "parameter_group": self._get_param("DBClusterParameterGroupName"),
-            "region": self.region,
             "db_cluster_instance_class": self._get_param("DBClusterInstanceClass"),
             "enable_http_endpoint": self._get_bool_param("EnableHttpEndpoint"),
             "copy_tags_to_snapshot": self._get_bool_param("CopyTagsToSnapshot"),
diff --git a/moto/settings.py b/moto/settings.py
index a537d0143..09735052c 100644
--- a/moto/settings.py
+++ b/moto/settings.py
@@ -191,3 +191,11 @@ def load_iam_aws_managed_policies() -> bool:
         is True
         or os.environ.get("MOTO_IAM_LOAD_MANAGED_POLICIES", "").lower() == "true"
     )
+
+
+#
+# NOTE:  Recommend the next major release to set this to True for proper
+# default behavior
+#
+def iot_use_valid_cert() -> bool:
+    return default_user_config.get("iot", {}).get("use_valid_cert", False)
diff --git a/moto/ssm/resources/regions.json b/moto/ssm/resources/regions.json
index 74da42b47..68f2469fc 100644
--- a/moto/ssm/resources/regions.json
+++ b/moto/ssm/resources/regions.json
@@ -3663,7 +3663,13 @@
                 }
               },
               "bedrock": {
-                "Value": "bedrock"
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.ap-northeast-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "budgets": {
                 "Value": "budgets",
@@ -8420,6 +8426,15 @@
               "vpc": {
                 "Value": "vpc"
               },
+              "vpc-lattice": {
+                "Value": "vpc-lattice",
+                "endpoint": {
+                  "Value": "vpc-lattice.ap-northeast-2.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "vpn": {
                 "Value": "vpn",
                 "endpoint": {
@@ -10407,7 +10422,13 @@
                 }
               },
               "bedrock": {
-                "Value": "bedrock"
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.ap-south-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "budgets": {
                 "Value": "budgets",
@@ -14020,7 +14041,13 @@
                 }
               },
               "bedrock": {
-                "Value": "bedrock"
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.ap-southeast-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "budgets": {
                 "Value": "budgets",
@@ -16651,7 +16678,13 @@
                 }
               },
               "bedrock": {
-                "Value": "bedrock"
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.ap-southeast-2.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "budgets": {
                 "Value": "budgets",
@@ -22086,6 +22119,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "bedrock": {
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.ca-central-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "budgets": {
                 "Value": "budgets",
                 "endpoint": {
@@ -24693,6 +24735,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "identity-center": {
+                "Value": "identity-center",
+                "endpoint": {
+                  "Value": "sso.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "imagebuilder": {
                 "Value": "imagebuilder",
                 "endpoint": {
@@ -25086,6 +25137,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "sso-oidc": {
+                "Value": "sso-oidc",
+                "endpoint": {
+                  "Value": "oidc.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "stepfunctions": {
                 "Value": "stepfunctions",
                 "endpoint": {
@@ -28305,7 +28365,13 @@
                 }
               },
               "bedrock": {
-                "Value": "bedrock"
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.eu-central-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "budgets": {
                 "Value": "budgets",
@@ -31748,6 +31814,9 @@
                   "Value": "HTTPS"
                 }
               },
+              "rosa": {
+                "Value": "rosa"
+              },
               "route53": {
                 "Value": "route53",
                 "endpoint": {
@@ -37312,7 +37381,13 @@
                 }
               },
               "bedrock": {
-                "Value": "bedrock"
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.eu-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "budgets": {
                 "Value": "budgets",
@@ -40144,6 +40219,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "bedrock": {
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.eu-west-2.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "braket": {
                 "Value": "braket",
                 "endpoint": {
@@ -42650,7 +42734,13 @@
                 }
               },
               "bedrock": {
-                "Value": "bedrock"
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.eu-west-3.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "budgets": {
                 "Value": "budgets",
@@ -49052,6 +49142,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "bedrock": {
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.sa-east-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "budgets": {
                 "Value": "budgets",
                 "endpoint": {
@@ -51183,7 +51282,13 @@
                 }
               },
               "bedrock": {
-                "Value": "bedrock"
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.us-east-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "billingconductor": {
                 "Value": "billingconductor",
@@ -61745,6 +61850,15 @@
               "vpc": {
                 "Value": "vpc"
               },
+              "vpc-lattice": {
+                "Value": "vpc-lattice",
+                "endpoint": {
+                  "Value": "vpc-lattice.us-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "vpn": {
                 "Value": "vpn",
                 "endpoint": {
@@ -62145,7 +62259,13 @@
                 }
               },
               "bedrock": {
-                "Value": "bedrock"
+                "Value": "bedrock",
+                "endpoint": {
+                  "Value": "bedrock.us-west-2.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "braket": {
                 "Value": "braket",
diff --git a/moto/ssm/resources/services.json b/moto/ssm/resources/services.json
index 3ec21c830..199517864 100644
--- a/moto/ssm/resources/services.json
+++ b/moto/ssm/resources/services.json
@@ -7571,34 +7571,115 @@
             },
             "regions": {
               "ap-northeast-1": {
-                "Value": "ap-northeast-1"
+                "Value": "ap-northeast-1",
+                "endpoint": {
+                  "Value": "bedrock.ap-northeast-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "ap-south-1": {
-                "Value": "ap-south-1"
+                "Value": "ap-south-1",
+                "endpoint": {
+                  "Value": "bedrock.ap-south-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "ap-southeast-1": {
-                "Value": "ap-southeast-1"
+                "Value": "ap-southeast-1",
+                "endpoint": {
+                  "Value": "bedrock.ap-southeast-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "ap-southeast-2": {
-                "Value": "ap-southeast-2"
+                "Value": "ap-southeast-2",
+                "endpoint": {
+                  "Value": "bedrock.ap-southeast-2.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
+              "ca-central-1": {
+                "Value": "ca-central-1",
+                "endpoint": {
+                  "Value": "bedrock.ca-central-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "eu-central-1": {
-                "Value": "eu-central-1"
+                "Value": "eu-central-1",
+                "endpoint": {
+                  "Value": "bedrock.eu-central-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "eu-west-1": {
-                "Value": "eu-west-1"
+                "Value": "eu-west-1",
+                "endpoint": {
+                  "Value": "bedrock.eu-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
+              "eu-west-2": {
+                "Value": "eu-west-2",
+                "endpoint": {
+                  "Value": "bedrock.eu-west-2.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "eu-west-3": {
-                "Value": "eu-west-3"
+                "Value": "eu-west-3",
+                "endpoint": {
+                  "Value": "bedrock.eu-west-3.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
+              "sa-east-1": {
+                "Value": "sa-east-1",
+                "endpoint": {
+                  "Value": "bedrock.sa-east-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "us-east-1": {
-                "Value": "us-east-1"
+                "Value": "us-east-1",
+                "endpoint": {
+                  "Value": "bedrock.us-east-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               },
               "us-gov-west-1": {
                 "Value": "us-gov-west-1"
               },
               "us-west-2": {
-                "Value": "us-west-2"
+                "Value": "us-west-2",
+                "endpoint": {
+                  "Value": "bedrock.us-west-2.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
               }
             }
           },
@@ -8200,7 +8281,7 @@
           "chime-sdk-meetings": {
             "Value": "chime-sdk-meetings",
             "longName": {
-              "Value": "AWS Chime Meetings SDK"
+              "Value": "Amazon Chime SDK Meetings"
             },
             "regions": {
               "ap-northeast-1": {
@@ -30107,6 +30188,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "sso.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "cn-north-1": {
                 "Value": "cn-north-1",
                 "endpoint": {
@@ -41461,7 +41551,7 @@
           "memorydb": {
             "Value": "memorydb",
             "longName": {
-              "Value": "Amazon MemoryDB for Redis"
+              "Value": "Amazon MemoryDB"
             },
             "marketingHomeURL": {
               "Value": "https://aws.amazon.com/memorydb"
@@ -46888,11 +46978,12 @@
             }
           },
           "pinpoint-sms-voice-v2": {
+            "Value": "pinpoint-sms-voice-v2",
             "longName": {
-              "Value": "Amazon Pinpoint SMS Voice V2"
+              "Value": "AWS End User Messaging"
             },
             "marketingHomeURL": {
-              "Value": "https://aws.amazon.com/pinpoint/"
+              "Value": "https://aws.amazon.com/end-user-messaging"
             },
             "regions": {
               "af-south-1": {
@@ -51905,6 +51996,9 @@
               "eu-central-1": {
                 "Value": "eu-central-1"
               },
+              "eu-central-2": {
+                "Value": "eu-central-2"
+              },
               "eu-north-1": {
                 "Value": "eu-north-1"
               },
@@ -60824,6 +60918,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ca-west-1": {
+                "Value": "ca-west-1",
+                "endpoint": {
+                  "Value": "oidc.ca-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "cn-north-1": {
                 "Value": "cn-north-1",
                 "endpoint": {
@@ -65288,6 +65391,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "ap-northeast-2": {
+                "Value": "ap-northeast-2",
+                "endpoint": {
+                  "Value": "vpc-lattice.ap-northeast-2.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "ap-southeast-1": {
                 "Value": "ap-southeast-1",
                 "endpoint": {
@@ -65378,6 +65490,15 @@
                   "Value": "HTTPS"
                 }
               },
+              "us-west-1": {
+                "Value": "us-west-1",
+                "endpoint": {
+                  "Value": "vpc-lattice.us-west-1.amazonaws.com"
+                },
+                "protocols": {
+                  "Value": "HTTPS"
+                }
+              },
               "us-west-2": {
                 "Value": "us-west-2",
                 "endpoint": {


Test patch:
diff --git a/tests/test_dynamodb/test_dynamodb_export_table.py b/tests/test_dynamodb/test_dynamodb_export_table.py
new file mode 100644
index 000000000..7ee25dc5b
--- /dev/null
+++ b/tests/test_dynamodb/test_dynamodb_export_table.py
@@ -0,0 +1,361 @@
+import gzip
+import json
+from time import sleep
+from unittest import SkipTest, mock
+from uuid import uuid4
+
+import boto3
+import pytest
+
+from moto import settings
+from moto.dynamodb.models import TableExport
+
+from . import dynamodb_aws_verified
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_export_from_missing_table(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-1")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    s3_prefix = "prefix"
+    s3.create_bucket(Bucket=s3_bucket_name)
+    table_arn = "t" + str(uuid4())[0:6]
+
+    export_description = client.export_table_to_point_in_time(
+        TableArn=table_arn,
+        ExportFormat="DYNAMO_JSON",
+        S3Bucket=s3_bucket_name,
+        S3Prefix=s3_prefix,
+    )["ExportDescription"]
+
+    export_details = wait_for_export(client, export_description)
+
+    assert export_details["ExportStatus"] == "FAILED"
+    assert export_details["FailureCode"] == "DynamoDBTableNotFound"
+    assert "The specified table does not exist" in export_details["FailureMessage"]
+
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_export_to_missing_s3_bucket(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    s3_prefix = "prefix"
+    table_name = "test-table"
+
+    response = client.create_table(
+        TableName=table_name,
+        KeySchema=[{"AttributeName": "username", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "username", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+    table_arn = response["TableDescription"]["TableArn"]
+
+    client.update_continuous_backups(
+        TableName=table_name,
+        PointInTimeRecoverySpecification={"PointInTimeRecoveryEnabled": True},
+    )
+
+    export_description = client.export_table_to_point_in_time(
+        TableArn=table_arn,
+        ExportFormat="DYNAMO_JSON",
+        S3Bucket=s3_bucket_name,
+        S3Prefix=s3_prefix,
+    )["ExportDescription"]
+
+    export_details = wait_for_export(client, export_description)
+
+    assert export_details["ExportStatus"] == "FAILED"
+    assert export_details["FailureCode"] == "S3NoSuchBucket"
+    assert "The specified bucket does not exist" in export_details["FailureMessage"]
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_export_point_in_time_recovery_not_enabled(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    s3_prefix = "prefix"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+    response = client.create_table(
+        TableName=table_name,
+        KeySchema=[{"AttributeName": "username", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "username", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+
+    table_arn = response["TableDescription"]["TableArn"]
+
+    export_description = client.export_table_to_point_in_time(
+        TableArn=table_arn,
+        ExportFormat="DYNAMO_JSON",
+        S3Bucket=s3_bucket_name,
+        S3Prefix=s3_prefix,
+    )["ExportDescription"]
+
+    export_details = wait_for_export(client, export_description)
+
+    assert export_details["ExportStatus"] == "FAILED"
+    assert export_details["FailureCode"] == "PointInTimeRecoveryUnavailable"
+    assert (
+        "Point in time recovery not enabled for table"
+        in export_details["FailureMessage"]
+    )
+
+    client.delete_table(TableName=table_name)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+@mock.patch.object(TableExport, "_backup_to_s3_file", mock.Mock(side_effect=Exception))
+def test_export_backup_to_s3_error(table_name=None):
+    if not settings.TEST_DECORATOR_MODE:
+        raise SkipTest("Can't mock backup to s3 error in server mode")
+
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    s3_prefix = "prefix"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+    response = client.create_table(
+        TableName=table_name,
+        KeySchema=[{"AttributeName": "username", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "username", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+    table_arn = response["TableDescription"]["TableArn"]
+    client.update_continuous_backups(
+        TableName=table_name,
+        PointInTimeRecoverySpecification={"PointInTimeRecoveryEnabled": True},
+    )
+
+    export_description = client.export_table_to_point_in_time(
+        TableArn=table_arn,
+        ExportFormat="DYNAMO_JSON",
+        S3Bucket=s3_bucket_name,
+        S3Prefix=s3_prefix,
+    )["ExportDescription"]
+
+    export_details = wait_for_export(client, export_description)
+
+    assert export_details["ExportStatus"] == "FAILED"
+    assert export_details["FailureCode"] == "UNKNOWN"
+
+    client.delete_table(TableName=table_name)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_export_empty_table(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    s3_prefix = "prefix"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+    response = client.create_table(
+        TableName=table_name,
+        KeySchema=[{"AttributeName": "username", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "username", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+    table_arn = response["TableDescription"]["TableArn"]
+
+    client.update_continuous_backups(
+        TableName=table_name,
+        PointInTimeRecoverySpecification={"PointInTimeRecoveryEnabled": True},
+    )
+
+    export_description = client.export_table_to_point_in_time(
+        TableArn=table_arn,
+        ExportFormat="DYNAMO_JSON",
+        S3Bucket=s3_bucket_name,
+        S3Prefix=s3_prefix,
+    )["ExportDescription"]
+
+    export_details = wait_for_export(client, export_description)
+
+    assert export_details["BilledSizeBytes"] == 0
+    assert export_details["ExportStatus"] == "COMPLETED"
+    assert export_details["ItemCount"] == 0
+    assert export_details["ExportFormat"] == "DYNAMO_JSON"
+    assert export_details["ExportType"] == "FULL_EXPORT"
+
+    s3_contents = s3.list_objects(Bucket=s3_bucket_name, Prefix=s3_prefix)["Contents"][
+        0
+    ]
+    object_key = s3_contents["Key"]
+    compressed_backup = s3.get_object(Bucket=s3_bucket_name, Key=object_key)[
+        "Body"
+    ].read()
+    file_contents = gzip.decompress(compressed_backup).decode("utf-8")
+    assert json.loads(file_contents) == []
+
+    client.delete_table(TableName=table_name)
+    s3.delete_object(Bucket=s3_bucket_name, Key=object_key)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_export_table(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    s3_prefix = "prefix"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+    response = client.create_table(
+        TableName=table_name,
+        KeySchema=[{"AttributeName": "username", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "username", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+    table_arn = response["TableDescription"]["TableArn"]
+
+    client.update_continuous_backups(
+        TableName=table_name,
+        PointInTimeRecoverySpecification={"PointInTimeRecoveryEnabled": True},
+    )
+
+    client.put_item(
+        TableName=table_name,
+        Item={"username": {"S": "user1"}, "binaryfoo": {"B": b"bar"}},
+    )
+    client.put_item(
+        TableName=table_name, Item={"username": {"S": "user2"}, "foo": {"S": "bar"}}
+    )
+    client.put_item(
+        TableName=table_name, Item={"username": {"S": "user3"}, "foo": {"S": "bar"}}
+    )
+
+    export_description = client.export_table_to_point_in_time(
+        TableArn=table_arn,
+        ExportFormat="DYNAMO_JSON",
+        S3Bucket=s3_bucket_name,
+        S3Prefix=s3_prefix,
+    )["ExportDescription"]
+
+    export_details = wait_for_export(client, export_description)
+    assert export_details["BilledSizeBytes"] == 3
+    assert export_details["ExportStatus"] == "COMPLETED"
+    assert export_details["ItemCount"] == 3
+    assert export_details["ExportFormat"] == "DYNAMO_JSON"
+    assert export_details["ExportType"] == "FULL_EXPORT"
+
+    s3_contents = s3.list_objects(Bucket=s3_bucket_name, Prefix=s3_prefix)["Contents"][
+        0
+    ]
+    object_key = s3_contents["Key"]
+    compressed_backup = s3.get_object(Bucket=s3_bucket_name, Key=object_key)[
+        "Body"
+    ].read()
+    file_contents = gzip.decompress(compressed_backup).decode("utf-8")
+    assert json.loads(file_contents) == [
+        {"Attributes": {"username": {"S": "user1"}, "binaryfoo": {"B": "YmFy"}}},
+        {"Attributes": {"username": {"S": "user2"}, "foo": {"S": "bar"}}},
+        {"Attributes": {"username": {"S": "user3"}, "foo": {"S": "bar"}}},
+    ]
+
+    client.delete_table(TableName=table_name)
+    s3.delete_object(Bucket=s3_bucket_name, Key=object_key)
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+@pytest.mark.aws_verified
+@dynamodb_aws_verified(create_table=False)
+def test_list_exports(table_name=None):
+    client = boto3.client("dynamodb", region_name="us-east-2")
+    s3 = boto3.client("s3", region_name="us-east-1")
+
+    s3_bucket_name = f"inttest{uuid4()}"
+    s3_prefix = "prefix"
+    table_name = "moto_test_" + str(uuid4())[0:6]
+
+    s3.create_bucket(Bucket=s3_bucket_name)
+    response = client.create_table(
+        TableName=table_name,
+        KeySchema=[{"AttributeName": "username", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "username", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+    table_arn = response["TableDescription"]["TableArn"]
+
+    client.update_continuous_backups(
+        TableName=table_name,
+        PointInTimeRecoverySpecification={"PointInTimeRecoveryEnabled": True},
+    )
+
+    client.put_item(
+        TableName=table_name,
+        Item={"username": {"S": "user1"}, "binaryfoo": {"B": b"bar"}},
+    )
+    client.put_item(
+        TableName=table_name, Item={"username": {"S": "user2"}, "foo": {"S": "bar"}}
+    )
+    client.put_item(
+        TableName=table_name, Item={"username": {"S": "user3"}, "foo": {"S": "bar"}}
+    )
+
+    export_description = client.export_table_to_point_in_time(
+        TableArn=table_arn,
+        ExportFormat="DYNAMO_JSON",
+        S3Bucket=s3_bucket_name,
+        S3Prefix=s3_prefix,
+    )["ExportDescription"]
+
+    export_details_1 = wait_for_export(client, export_description)
+    assert export_details_1["ExportStatus"] == "COMPLETED"
+    export_description = client.export_table_to_point_in_time(
+        TableArn=table_arn,
+        ExportFormat="DYNAMO_JSON",
+        S3Bucket=s3_bucket_name,
+        S3Prefix=s3_prefix,
+    )["ExportDescription"]
+
+    export_details_2 = wait_for_export(client, export_description)
+    assert export_details_2["ExportStatus"] == "COMPLETED"
+    export_summaries = client.list_exports(TableArn=table_arn)["ExportSummaries"]
+    assert len(export_summaries) == 2
+    assert export_summaries[0]["ExportStatus"] == "COMPLETED"
+    assert export_summaries[0]["ExportType"] == "FULL_EXPORT"
+    assert export_summaries[0]["ExportArn"] == export_details_1["ExportArn"]
+    assert export_summaries[1]["ExportStatus"] == "COMPLETED"
+    assert export_summaries[1]["ExportType"] == "FULL_EXPORT"
+    assert export_summaries[1]["ExportArn"] == export_details_2["ExportArn"]
+
+    client.delete_table(TableName=table_name)
+    s3_contents = s3.list_objects(Bucket=s3_bucket_name, Prefix=s3_prefix)["Contents"]
+    for object in s3_contents:
+        s3.delete_object(Bucket=s3_bucket_name, Key=object["Key"])
+    s3.delete_bucket(Bucket=s3_bucket_name)
+
+
+def wait_for_export(client, export_description):
+    status = "IN_PROGRESS"
+    while status == "IN_PROGRESS":
+        export_details = client.describe_export(
+            ExportArn=export_description["ExportArn"]
+        )
+        status = export_details["ExportDescription"]["ExportStatus"]
+        sleep(0.1)
+    return export_details["ExportDescription"]
diff --git a/tests/test_firehose/test_firehose_put.py b/tests/test_firehose/test_firehose_put.py
index 7a8170ae1..87f3e4385 100644
--- a/tests/test_firehose/test_firehose_put.py
+++ b/tests/test_firehose/test_firehose_put.py
@@ -1,5 +1,7 @@
 """Unit tests verifying put-related delivery stream APIs."""
 
+import re
+
 import boto3
 
 from moto import mock_aws
@@ -143,3 +145,6 @@ def test_put_record_batch_extended_s3_destination():
         Bucket=bucket_name, Key=bucket_objects["Contents"][0]["Key"]
     )
     assert response["Body"].read() == b"onetwothree"
+    assert re.match(
+        r"^[0-9]{4}/[0-9]{2}/[0-9]{2}/[0-9]{2}/", bucket_objects["Contents"][0]["Key"]
+    )
diff --git a/tests/test_iot/test_iot_ca_certificates.py b/tests/test_iot/test_iot_ca_certificates.py
index 9b75b7be7..ec8e41566 100644
--- a/tests/test_iot/test_iot_ca_certificates.py
+++ b/tests/test_iot/test_iot_ca_certificates.py
@@ -1,12 +1,17 @@
+from unittest import SkipTest
+
 import boto3
 import pytest
 from botocore.exceptions import ClientError
 
-from moto import mock_aws
+from moto import mock_aws, settings
 
 
 @mock_aws
-def test_register_ca_certificate_simple():
+def test_register_ca_certificate_simple_invalid_cert():
+    """
+    Original test case with invalid cert.
+    """
     client = boto3.client("iot", region_name="us-east-1")
 
     resp = client.register_ca_certificate(
@@ -18,6 +23,38 @@ def test_register_ca_certificate_simple():
     assert "certificateId" in resp
 
 
+@pytest.fixture(name="make_cert")
+@mock_aws(config={"iot": {"use_valid_cert": True}})
+def fixture_make_cert():
+    """
+    Create a valid cert for testing
+    """
+    client = boto3.client("iot", region_name="us-east-1")
+    return client.create_keys_and_certificate(setAsActive=False)
+
+
+@mock_aws(config={"iot": {"use_valid_cert": True}})
+def test_register_ca_certificate_simple(make_cert):
+    if not settings.TEST_DECORATOR_MODE:
+        # Technically it doesn't have to be skipped
+        # in ServerMode the config is not set, so we'll compute the cert the old/invalid way
+        # But there are no assertions to actually verify the value of the computed cert
+        # This is just added here as a warning/reminder if this ever changes
+        raise SkipTest("Config cannot be easily set in ServerMode")
+
+    client = boto3.client("iot", region_name="us-east-1")
+    certInfo = make_cert
+
+    resp = client.register_ca_certificate(
+        caCertificate=certInfo["certificatePem"],
+        verificationCertificate="verification_certificate",
+    )
+
+    assert "certificateArn" in resp
+    assert "certificateId" in resp
+    assert resp["certificateId"] == certInfo["certificateId"]
+
+
 @mock_aws
 def test_describe_ca_certificate_unknown():
     client = boto3.client("iot", region_name="us-east-2")
@@ -76,6 +113,10 @@ def test_describe_ca_certificate_advanced():
 
 @mock_aws
 def test_list_certificates_by_ca():
+    """
+    test listing certificates by CA.  Don't use valid cert to make it simpler
+    for the test.
+    """
     client = boto3.client("iot", region_name="us-east-1")
 
     # create ca
diff --git a/tests/test_rds/test_db_cluster_param_group.py b/tests/test_rds/test_db_cluster_param_group.py
index e1ceaf7e2..c3b753618 100644
--- a/tests/test_rds/test_db_cluster_param_group.py
+++ b/tests/test_rds/test_db_cluster_param_group.py
@@ -24,7 +24,7 @@ def test_create_describe_delete():
     assert group["Description"] == "familia"
     assert (
         group["DBClusterParameterGroupArn"]
-        == f"arn:aws:rds:us-east-2:{DEFAULT_ACCOUNT_ID}:cpg:groupname"
+        == f"arn:aws:rds:us-east-2:{DEFAULT_ACCOUNT_ID}:cluster-pg:groupname"
     )
 
     groups = client.describe_db_cluster_parameter_groups(
@@ -37,7 +37,7 @@ def test_create_describe_delete():
     assert groups[0]["Description"] == "familia"
     assert (
         groups[0]["DBClusterParameterGroupArn"]
-        == f"arn:aws:rds:us-east-2:{DEFAULT_ACCOUNT_ID}:cpg:groupname"
+        == f"arn:aws:rds:us-east-2:{DEFAULT_ACCOUNT_ID}:cluster-pg:groupname"
     )
 
     client.delete_db_cluster_parameter_group(DBClusterParameterGroupName="groupname")
diff --git a/tests/test_rds/test_rds_proxy.py b/tests/test_rds/test_rds_proxy.py
index c3d76fd7b..602289a5e 100644
--- a/tests/test_rds/test_rds_proxy.py
+++ b/tests/test_rds/test_rds_proxy.py
@@ -43,7 +43,7 @@ def test_create_db_proxy():
         db_proxy["DBProxyArn"]
         == f"arn:aws:rds:us-west-2:{ACCOUNT_ID}:db-proxy:testrdsproxy"
     )
-    assert db_proxy["Status"] == "availible"
+    assert db_proxy["Status"] == "available"
     assert db_proxy["EngineFamily"] == "MYSQL"
     assert db_proxy["VpcId"] == vpc_id
     assert db_proxy["VpcSecurityGroupIds"] == []
@@ -103,7 +103,7 @@ def test_describe_db_proxies():
         db_proxy["DBProxyArn"]
         == f"arn:aws:rds:us-west-2:{ACCOUNT_ID}:db-proxy:testrdsproxydescribe"
     )
-    assert db_proxy["Status"] == "availible"
+    assert db_proxy["Status"] == "available"
     assert db_proxy["EngineFamily"] == "MYSQL"
     assert db_proxy["VpcId"] == vpc_id
     assert db_proxy["VpcSecurityGroupIds"] == []