Repository: getmoto/moto
Pull request #7716 (2024-05-22 09:05:20)

Partitions: Regex validation and missing ARN's
Related to #7699 


Patch:
diff --git a/moto/acmpca/models.py b/moto/acmpca/models.py
index 6a2057ee5..ab260be2c 100644
--- a/moto/acmpca/models.py
+++ b/moto/acmpca/models.py
@@ -15,6 +15,7 @@ from moto.core.common_models import BaseModel
 from moto.core.utils import unix_time, utcnow
 from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 from .exceptions import InvalidS3ObjectAclInCrlConfiguration, ResourceNotFoundException
 
@@ -30,9 +31,7 @@ class CertificateAuthority(BaseModel):
         security_standard: Optional[str],
     ):
         self.id = mock_random.uuid4()
-        self.arn = (
-            f"arn:aws:acm-pca:{region}:{account_id}:certificate-authority/{self.id}"
-        )
+        self.arn = f"arn:{get_partition(region)}:acm-pca:{region}:{account_id}:certificate-authority/{self.id}"
         self.account_id = account_id
         self.region_name = region
         self.certificate_authority_configuration = certificate_authority_configuration
@@ -119,7 +118,7 @@ class CertificateAuthority(BaseModel):
         cert = cryptography.x509.load_pem_x509_csr(base64.b64decode(csr_bytes))
         new_cert = self.generate_cert(common_name="", subject=cert.subject)
         cert_id = str(mock_random.uuid4()).replace("-", "")
-        cert_arn = f"arn:aws:acm-pca:{self.region_name}:{self.account_id}:certificate-authority/{self.id}/certificate/{cert_id}"
+        cert_arn = f"arn:{get_partition(self.region_name)}:acm-pca:{self.region_name}:{self.account_id}:certificate-authority/{self.id}/certificate/{cert_id}"
         self.issued_certificates[cert_arn] = new_cert
         return cert_arn
 
diff --git a/moto/apigateway/models.py b/moto/apigateway/models.py
index 0c2c29b9c..9ca56026a 100644
--- a/moto/apigateway/models.py
+++ b/moto/apigateway/models.py
@@ -23,6 +23,7 @@ from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel, CloudFormationModel
 from moto.core.utils import path_url
 from moto.moto_api._internal import mock_random as random
+from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition
 
 from ..core.models import responses_mock
 from .exceptions import (
@@ -1950,7 +1951,8 @@ class APIGatewayBackend(BaseBackend):
     ) -> Integration:
         resource = self.get_resource(function_id, resource_id)
         if credentials and not re.match(
-            "^arn:aws:iam::" + str(self.account_id), credentials
+            f"^arn:{get_partition(self.region_name)}:iam::" + str(self.account_id),
+            credentials,
         ):
             raise CrossAccountNotAllowed()
         if not integration_method and integration_type in [
@@ -1961,21 +1963,25 @@ class APIGatewayBackend(BaseBackend):
         ]:
             raise IntegrationMethodNotDefined()
         if integration_type in ["AWS_PROXY"] and re.match(
-            "^arn:aws:apigateway:[a-zA-Z0-9-]+:s3", uri
+            ARN_PARTITION_REGEX + ":apigateway:[a-zA-Z0-9-]+:s3", uri
         ):
             raise AwsProxyNotAllowed()
         if (
             integration_type in ["AWS"]
-            and re.match("^arn:aws:apigateway:[a-zA-Z0-9-]+:s3", uri)
+            and re.match(ARN_PARTITION_REGEX + ":apigateway:[a-zA-Z0-9-]+:s3", uri)
             and not credentials
         ):
             raise RoleNotSpecified()
         if integration_type in ["HTTP", "HTTP_PROXY"] and not self._uri_validator(uri):
             raise InvalidHttpEndpoint()
-        if integration_type in ["AWS", "AWS_PROXY"] and not re.match("^arn:aws:", uri):
+        if integration_type in ["AWS", "AWS_PROXY"] and not re.match(
+            ARN_PARTITION_REGEX + ":", uri
+        ):
             raise InvalidArn()
         if integration_type in ["AWS", "AWS_PROXY"] and not re.match(
-            "^arn:aws:apigateway:[a-zA-Z0-9-]+:[a-zA-Z0-9-.]+:(path|action)/", uri
+            ARN_PARTITION_REGEX
+            + ":apigateway:[a-zA-Z0-9-]+:[a-zA-Z0-9-.]+:(path|action)/",
+            uri,
         ):
             raise InvalidIntegrationArn()
         integration = resource.add_integration(
diff --git a/moto/apigateway/urls.py b/moto/apigateway/urls.py
index 1b520f313..03567ddd2 100644
--- a/moto/apigateway/urls.py
+++ b/moto/apigateway/urls.py
@@ -11,7 +11,7 @@ url_paths = {
     "{0}/restapis/(?P<function_id>[^/]+)/authorizers/(?P<authorizer_id>[^/]+)/?$": APIGatewayResponse.dispatch,
     "{0}/restapis/(?P<function_id>[^/]+)/stages$": APIGatewayResponse.dispatch,
     "{0}/tags/(?P<resourceArn>[^/]+)$": APIGatewayResponse.dispatch,
-    "{0}/tags/arn:aws:apigateway:(?P<region_name>[^/]+)::/restapis/(?P<function_id>[^/]+)/stages/(?P<stage_name>[^/]+)$": APIGatewayResponse.dispatch,
+    "{0}/tags/arn:(?P<partition>[^/]+):apigateway:(?P<region_name>[^/]+)::/restapis/(?P<function_id>[^/]+)/stages/(?P<stage_name>[^/]+)$": APIGatewayResponse.dispatch,
     "{0}/restapis/(?P<function_id>[^/]+)/stages/(?P<stage_name>[^/]+)/?$": APIGatewayResponse.dispatch,
     "{0}/restapis/(?P<function_id>[^/]+)/stages/(?P<stage_name>[^/]+)/exports/(?P<export_type>[^/]+)/?$": APIGatewayResponse.dispatch,
     "{0}/restapis/(?P<function_id>[^/]+)/deployments$": APIGatewayResponse.dispatch,
diff --git a/moto/applicationautoscaling/models.py b/moto/applicationautoscaling/models.py
index acddc7706..d3d2a1c64 100644
--- a/moto/applicationautoscaling/models.py
+++ b/moto/applicationautoscaling/models.py
@@ -1,3 +1,4 @@
+import re
 import time
 from collections import OrderedDict
 from enum import Enum, unique
@@ -7,7 +8,7 @@ from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.ecs import ecs_backends
 from moto.moto_api._internal import mock_random
-from moto.utilities.utils import get_partition
+from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition
 
 from .exceptions import AWSValidationException
 
@@ -379,7 +380,7 @@ def _get_resource_type_from_resource_id(resource_id: str) -> str:
     #    - ...except for sagemaker endpoints, dynamodb GSIs and keyspaces tables, where it's the third.
     #  - Comprehend uses an arn, with the resource type being the last element.
 
-    if resource_id.startswith("arn:aws:comprehend"):
+    if re.match(ARN_PARTITION_REGEX + ":comprehend", resource_id):
         resource_id = resource_id.split(":")[-1]
     resource_split = (
         resource_id.split("/") if "/" in resource_id else resource_id.split(":")
diff --git a/moto/appsync/models.py b/moto/appsync/models.py
index fe9970028..f0e0345f6 100644
--- a/moto/appsync/models.py
+++ b/moto/appsync/models.py
@@ -41,8 +41,9 @@ directive @aws_cognito_user_pools(
 
 
 class GraphqlSchema(BaseModel):
-    def __init__(self, definition: Any):
+    def __init__(self, definition: Any, region_name: str):
         self.definition = definition
+        self.region_name = region_name
         # [graphql.language.ast.ObjectTypeDefinitionNode, ..]
         self.types: List[Any] = []
 
@@ -58,7 +59,7 @@ class GraphqlSchema(BaseModel):
                     "description": graphql_type.description.value
                     if graphql_type.description
                     else None,
-                    "arn": f"arn:aws:appsync:graphql_type/{name}",
+                    "arn": f"arn:{get_partition(self.region_name)}:appsync:graphql_type/{name}",
                     "definition": "NotYetImplemented",
                 }
 
@@ -212,7 +213,7 @@ class GraphqlAPI(BaseModel):
     def start_schema_creation(self, definition: str) -> None:
         graphql_definition = base64.b64decode(definition).decode("utf-8")
 
-        self.graphql_schema = GraphqlSchema(graphql_definition)
+        self.graphql_schema = GraphqlSchema(graphql_definition, region_name=self.region)
 
     def get_schema_status(self) -> Any:
         return self.graphql_schema.get_status()  # type: ignore[union-attr]
diff --git a/moto/autoscaling/models.py b/moto/autoscaling/models.py
index 3cd6371d1..c1265c795 100644
--- a/moto/autoscaling/models.py
+++ b/moto/autoscaling/models.py
@@ -449,7 +449,8 @@ class FakeAutoScalingGroup(CloudFormationModel):
         self._id = str(random.uuid4())
         self.region = self.autoscaling_backend.region_name
         self.account_id = self.autoscaling_backend.account_id
-        self.service_linked_role = f"arn:aws:iam::{self.account_id}:role/aws-service-role/autoscaling.amazonaws.com/AWSServiceRoleForAutoScaling"
+        partition = get_partition(self.region)
+        self.service_linked_role = f"arn:{partition}:iam::{self.account_id}:role/aws-service-role/autoscaling.amazonaws.com/AWSServiceRoleForAutoScaling"
 
         self.vpc_zone_identifier: Optional[str] = None
         self._set_azs_and_vpcs(availability_zones, vpc_zone_identifier)
diff --git a/moto/awslambda/models.py b/moto/awslambda/models.py
index 253f7ae71..ce66fef66 100644
--- a/moto/awslambda/models.py
+++ b/moto/awslambda/models.py
@@ -40,7 +40,11 @@ from moto.s3.exceptions import MissingBucket, MissingKey
 from moto.s3.models import FakeKey, s3_backends
 from moto.sqs.models import sqs_backends
 from moto.utilities.docker_utilities import DockerModel
-from moto.utilities.utils import get_partition, load_resource_as_bytes
+from moto.utilities.utils import (
+    ARN_PARTITION_REGEX,
+    get_partition,
+    load_resource_as_bytes,
+)
 
 from .exceptions import (
     ConflictException,
@@ -1606,7 +1610,7 @@ class LambdaStorage(object):
         :raises: InvalidParameterValue if qualifier is provided
         """
 
-        if name_or_arn.startswith("arn:aws"):
+        if re.match(ARN_PARTITION_REGEX, name_or_arn):
             [_, name, qualifier] = self.split_function_arn(name_or_arn)
 
             if qualifier is not None:
@@ -1625,7 +1629,7 @@ class LambdaStorage(object):
         :raises: UnknownFunctionException if function not found
         """
 
-        if name_or_arn.startswith("arn:aws"):
+        if re.match(ARN_PARTITION_REGEX, name_or_arn):
             [_, name, qualifier_in_arn] = self.split_function_arn(name_or_arn)
             return self.get_function_by_name_with_qualifier(
                 name, qualifier_in_arn or qualifier
@@ -1636,7 +1640,7 @@ class LambdaStorage(object):
     def construct_unknown_function_exception(
         self, name_or_arn: str, qualifier: Optional[str] = None
     ) -> UnknownFunctionException:
-        if name_or_arn.startswith("arn:aws"):
+        if re.match(ARN_PARTITION_REGEX, name_or_arn):
             arn = name_or_arn
         else:
             # name_or_arn is a function name with optional qualifier <func_name>[:<qualifier>]
@@ -1696,7 +1700,7 @@ class LambdaStorage(object):
 
     def del_function(self, name_or_arn: str, qualifier: Optional[str] = None) -> None:
         # Qualifier may be explicitly passed or part of function name or ARN, extract it here
-        if name_or_arn.startswith("arn:aws"):
+        if re.match(ARN_PARTITION_REGEX, name_or_arn):
             # Extract from ARN
             if ":" in name_or_arn.split(":function:")[-1]:
                 qualifier = name_or_arn.split(":")[-1]
diff --git a/moto/awslambda/responses.py b/moto/awslambda/responses.py
index af87165a2..4f825b991 100644
--- a/moto/awslambda/responses.py
+++ b/moto/awslambda/responses.py
@@ -1,10 +1,12 @@
 import json
+import re
 import sys
 from typing import Any, Dict, List, Tuple, Union
 from urllib.parse import unquote
 
 from moto.core.responses import TYPE_RESPONSE, BaseResponse
 from moto.utilities.aws_headers import amz_crc32
+from moto.utilities.utils import ARN_PARTITION_REGEX
 
 from .exceptions import FunctionAlreadyExists, UnknownFunctionException
 from .models import LambdaBackend
@@ -224,7 +226,7 @@ class LambdaResponse(BaseResponse):
         configuration: Dict[str, Any], function_name: str, qualifier: str
     ) -> Dict[str, Any]:
         # Qualifier may be explicitly passed or part of function name or ARN, extract it here
-        if function_name.startswith("arn:aws"):
+        if re.match(ARN_PARTITION_REGEX, function_name):
             # Extract from ARN
             if ":" in function_name.split(":function:")[-1]:
                 qualifier = function_name.split(":")[-1]
diff --git a/moto/awslambda/utils.py b/moto/awslambda/utils.py
index 003483af6..01ab0d85b 100644
--- a/moto/awslambda/utils.py
+++ b/moto/awslambda/utils.py
@@ -2,7 +2,7 @@ from collections import namedtuple
 from functools import partial
 from typing import TYPE_CHECKING, Any, Callable
 
-from moto.utilities.utils import get_partition
+from moto.utilities.utils import PARTITION_NAMES, get_partition
 
 if TYPE_CHECKING:
     from .models import LambdaBackend
@@ -33,7 +33,8 @@ make_layer_ver_arn = partial(make_ver_arn, "layer")
 
 
 def split_arn(arn_type: Callable[[str, str, str, str], str], arn: str) -> Any:
-    arn = arn.replace("arn:aws:lambda:", "")
+    for partition in PARTITION_NAMES:
+        arn = arn.replace(f"arn:{partition}:lambda:", "")
 
     region, account, _, name, version = arn.split(":")
 
diff --git a/moto/backend_index.py b/moto/backend_index.py
index 6406cc3bd..f57075621 100644
--- a/moto/backend_index.py
+++ b/moto/backend_index.py
@@ -21,12 +21,13 @@ backend_url_patterns = [
     ("awslambda", re.compile("https?://lambda\\.(.+)\\.amazonaws\\.com")),
     ("backup", re.compile("https?://backup\\.(.+)\\.amazonaws\\.com")),
     ("batch", re.compile("https?://batch\\.(.+)\\.amazonaws.com")),
-    ("budgets", re.compile("https?://budgets\\.amazonaws\\.com")),
     ("bedrock", re.compile("https?://bedrock\\.(.+)\\.amazonaws\\.com")),
     ("bedrockagent", re.compile("https?://bedrock-agent\\.(.+)\\.amazonaws\\.com")),
+    ("budgets", re.compile("https?://budgets\\.amazonaws\\.com")),
     ("ce", re.compile("https?://ce\\.(.+)\\.amazonaws\\.com")),
     ("cloudformation", re.compile("https?://cloudformation\\.(.+)\\.amazonaws\\.com")),
     ("cloudfront", re.compile("https?://cloudfront\\.amazonaws\\.com")),
+    ("cloudfront", re.compile("https?://cloudfront\\.(.+)\\.amazonaws\\.com")),
     ("cloudtrail", re.compile("https?://cloudtrail\\.(.+)\\.amazonaws\\.com")),
     ("cloudwatch", re.compile("https?://monitoring\\.(.+)\\.amazonaws.com")),
     ("codebuild", re.compile("https?://codebuild\\.(.+)\\.amazonaws\\.com")),
diff --git a/moto/ce/models.py b/moto/ce/models.py
index 354e1bf60..04245d82f 100644
--- a/moto/ce/models.py
+++ b/moto/ce/models.py
@@ -8,7 +8,7 @@ from moto.core.common_models import BaseModel
 from moto.core.utils import iso_8601_datetime_without_milliseconds
 from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
-from moto.utilities.utils import PARTITION_NAMES
+from moto.utilities.utils import PARTITION_NAMES, get_partition
 
 from .exceptions import CostCategoryNotFound
 
@@ -28,6 +28,7 @@ class CostCategoryDefinition(BaseModel):
     def __init__(
         self,
         account_id: str,
+        region_name: str,
         name: str,
         effective_start: Optional[str],
         rule_version: str,
@@ -40,7 +41,7 @@ class CostCategoryDefinition(BaseModel):
         self.rules = rules
         self.default_value = default_value
         self.split_charge_rules = split_charge_rules
-        self.arn = f"arn:aws:ce::{account_id}:costcategory/{str(mock_random.uuid4())}"
+        self.arn = f"arn:{get_partition(region_name)}:ce::{account_id}:costcategory/{str(mock_random.uuid4())}"
         self.effective_start: str = effective_start or first_day()
 
     def update(
@@ -93,13 +94,14 @@ class CostExplorerBackend(BaseBackend):
         The EffectiveOn and ResourceTags-parameters are not yet implemented
         """
         ccd = CostCategoryDefinition(
-            self.account_id,
-            name,
-            effective_start,
-            rule_version,
-            rules,
-            default_value,
-            split_charge_rules,
+            account_id=self.account_id,
+            region_name=self.region_name,
+            name=name,
+            effective_start=effective_start,
+            rule_version=rule_version,
+            rules=rules,
+            default_value=default_value,
+            split_charge_rules=split_charge_rules,
         )
         self.cost_categories[ccd.arn] = ccd
         self.tag_resource(ccd.arn, tags)
diff --git a/moto/cloudformation/models.py b/moto/cloudformation/models.py
index 73f7ed390..f9f8a2c2d 100644
--- a/moto/cloudformation/models.py
+++ b/moto/cloudformation/models.py
@@ -55,8 +55,10 @@ class FakeStackSet(BaseModel):
         self.description = description
         self.parameters = parameters
         self.tags = tags
-        self.admin_role = admin_role
-        self.admin_role_arn = f"arn:aws:iam::{account_id}:role/{self.admin_role}"
+        self.admin_role = (
+            admin_role
+            or f"arn:{get_partition(region)}:iam::{account_id}:role/AWSCloudFormationStackSetAdministrationRole"
+        )
         self.execution_role = execution_role or "AWSCloudFormationStackSetExecutionRole"
         self.status = "ACTIVE"
         self.instances = FakeStackInstances(
diff --git a/moto/cloudformation/responses.py b/moto/cloudformation/responses.py
index 528496713..66983ba65 100644
--- a/moto/cloudformation/responses.py
+++ b/moto/cloudformation/responses.py
@@ -531,8 +531,6 @@ class CloudFormationResponse(BaseResponse):
         stackset_name = self._get_param("StackSetName")
         stackset = self.cloudformation_backend.describe_stack_set(stackset_name)
 
-        if not stackset.admin_role:
-            stackset.admin_role = f"arn:aws:iam::{self.current_account}:role/AWSCloudFormationStackSetAdministrationRole"
         if not stackset.execution_role:
             stackset.execution_role = "AWSCloudFormationStackSetExecutionRole"
 
@@ -1208,7 +1206,7 @@ DESCRIBE_STACKSET_OPERATION_RESPONSE_TEMPLATE = """<DescribeStackSetOperationRes
   <DescribeStackSetOperationResult>
     <StackSetOperation>
       <ExecutionRoleName>{{ stackset.execution_role }}</ExecutionRoleName>
-      <AdministrationRoleARN>{{ stackset.admin_role_arn }}</AdministrationRoleARN>
+      <AdministrationRoleARN>{{ stackset.admin_role }}</AdministrationRoleARN>
       <StackSetId>{{ stackset.id }}</StackSetId>
       <CreationTimestamp>{{ operation.CreationTimestamp }}</CreationTimestamp>
       <OperationId>{{ operation.OperationId }}</OperationId>
diff --git a/moto/cloudfront/models.py b/moto/cloudfront/models.py
index d3f264a8a..c720c3d8f 100644
--- a/moto/cloudfront/models.py
+++ b/moto/cloudfront/models.py
@@ -7,7 +7,7 @@ from moto.core.utils import iso_8601_datetime_with_milliseconds
 from moto.moto_api._internal import mock_random as random
 from moto.moto_api._internal.managed_state_model import ManagedState
 from moto.utilities.tagging_service import TaggingService
-from moto.utilities.utils import PARTITION_NAMES
+from moto.utilities.utils import PARTITION_NAMES, get_partition
 
 from .exceptions import (
     DistributionAlreadyExists,
@@ -209,16 +209,14 @@ class Distribution(BaseModel, ManagedState):
         )
         return resource_id
 
-    def __init__(self, account_id: str, config: Dict[str, Any]):
+    def __init__(self, account_id: str, region_name: str, config: Dict[str, Any]):
         # Configured ManagedState
         super().__init__(
             "cloudfront::distribution", transitions=[("InProgress", "Deployed")]
         )
         # Configure internal properties
         self.distribution_id = Distribution.random_id()
-        self.arn = (
-            f"arn:aws:cloudfront:{account_id}:distribution/{self.distribution_id}"
-        )
+        self.arn = f"arn:{get_partition(region_name)}:cloudfront:{account_id}:distribution/{self.distribution_id}"
         self.distribution_config = DistributionConfig(config)
         self.active_trusted_signers = ActiveTrustedSigners()
         self.active_trusted_key_groups = ActiveTrustedKeyGroups()
@@ -306,7 +304,7 @@ class CloudFrontBackend(BaseBackend):
     def create_distribution_with_tags(
         self, distribution_config: Dict[str, Any], tags: List[Dict[str, str]]
     ) -> Tuple[Distribution, str, str]:
-        dist = Distribution(self.account_id, distribution_config)
+        dist = Distribution(self.account_id, self.region_name, distribution_config)
         caller_reference = dist.distribution_config.caller_reference
         existing_dist = self._distribution_with_caller_reference(caller_reference)
         if existing_dist is not None:
diff --git a/moto/cloudfront/urls.py b/moto/cloudfront/urls.py
index edf584fcf..a630a1f28 100644
--- a/moto/cloudfront/urls.py
+++ b/moto/cloudfront/urls.py
@@ -4,6 +4,7 @@ from .responses import CloudFrontResponse
 
 url_bases = [
     r"https?://cloudfront\.amazonaws\.com",
+    r"https?://cloudfront\.(.+)\.amazonaws\.com",
 ]
 url_paths = {
     "{0}/2020-05-31/distribution$": CloudFrontResponse.dispatch,
diff --git a/moto/cloudwatch/models.py b/moto/cloudwatch/models.py
index bb4f8ab95..20849ebfe 100644
--- a/moto/cloudwatch/models.py
+++ b/moto/cloudwatch/models.py
@@ -304,9 +304,9 @@ class MetricAggregatedDatum(MetricDatumBase):
 
 
 class Dashboard(BaseModel):
-    def __init__(self, account_id: str, name: str, body: str):
+    def __init__(self, account_id: str, region_name: str, name: str, body: str):
         # Guaranteed to be unique for now as the name is also the key of a dictionary where they are stored
-        self.arn = make_arn_for_dashboard(account_id, name)
+        self.arn = make_arn_for_dashboard(account_id, region_name, name)
         self.name = name
         self.body = body
         self.last_modified = datetime.now()
@@ -809,7 +809,7 @@ class CloudWatchBackend(BaseBackend):
         return self.metric_data + self.aws_metric_data
 
     def put_dashboard(self, name: str, body: str) -> None:
-        self.dashboards[name] = Dashboard(self.account_id, name, body)
+        self.dashboards[name] = Dashboard(self.account_id, self.region_name, name, body)
 
     def list_dashboards(self, prefix: str = "") -> Iterable[Dashboard]:
         for key, value in self.dashboards.items():
diff --git a/moto/cloudwatch/utils.py b/moto/cloudwatch/utils.py
index c106a9a76..539410446 100644
--- a/moto/cloudwatch/utils.py
+++ b/moto/cloudwatch/utils.py
@@ -1,8 +1,8 @@
 from moto.utilities.utils import get_partition
 
 
-def make_arn_for_dashboard(account_id: str, name: str) -> str:
-    return f"arn:aws:cloudwatch::{account_id}dashboard/{name}"
+def make_arn_for_dashboard(account_id: str, region_name: str, name: str) -> str:
+    return f"arn:{get_partition(region_name)}:cloudwatch::{account_id}:dashboard/{name}"
 
 
 def make_arn_for_alarm(region: str, account_id: str, alarm_name: str) -> str:
diff --git a/moto/codebuild/models.py b/moto/codebuild/models.py
index 9ac25c4b3..34b96311d 100644
--- a/moto/codebuild/models.py
+++ b/moto/codebuild/models.py
@@ -114,7 +114,7 @@ class CodeBuild(BaseModel):
             f"arn:{get_partition(region)}:kms:{region}:{account_id}:alias/aws/s3"
         )
         self.project_metadata["serviceRole"] = (
-            f"arn:aws:iam::{account_id}:role/service-role/{serviceRole}"
+            f"arn:{get_partition(region)}:iam::{account_id}:role/service-role/{serviceRole}"
         )
         self.project_metadata["lastModifiedDate"] = current_date
         self.project_metadata["created"] = current_date
diff --git a/moto/codebuild/responses.py b/moto/codebuild/responses.py
index bcdaecef5..f516f6c2c 100644
--- a/moto/codebuild/responses.py
+++ b/moto/codebuild/responses.py
@@ -3,6 +3,7 @@ import re
 from typing import Any, Dict, List
 
 from moto.core.responses import BaseResponse
+from moto.utilities.utils import get_partition
 
 from .exceptions import (
     InvalidInputException,
@@ -31,8 +32,12 @@ def _validate_required_params_source(source: Dict[str, Any]) -> None:
         raise InvalidInputException("Project source location is required")
 
 
-def _validate_required_params_service_role(account_id: str, service_role: str) -> None:
-    if not service_role.startswith(f"arn:aws:iam::{account_id}:role/"):
+def _validate_required_params_service_role(
+    account_id: str, region_name: str, service_role: str
+) -> None:
+    if not service_role.startswith(
+        f"arn:{get_partition(region_name)}:iam::{account_id}:role/"
+    ):
         raise InvalidInputException(
             "Invalid service role: Service role account ID does not match caller's account"
         )
@@ -105,7 +110,7 @@ class CodeBuildResponse(BaseResponse):
         ):
             name = self._get_param("projectName")
             raise ResourceNotFoundException(
-                f"The provided project arn:aws:codebuild:{self.region}:{self.current_account}:project/{name} does not exist"
+                f"The provided project arn:{get_partition(self.region)}:codebuild:{self.region}:{self.current_account}:project/{name} does not exist"
             )
 
         ids = self.codebuild_backend.list_builds_for_project(
@@ -117,7 +122,9 @@ class CodeBuildResponse(BaseResponse):
     def create_project(self) -> str:
         _validate_required_params_source(self._get_param("source"))
         service_role = self._get_param("serviceRole")
-        _validate_required_params_service_role(self.current_account, service_role)
+        _validate_required_params_service_role(
+            self.current_account, self.region, service_role
+        )
         _validate_required_params_artifacts(self._get_param("artifacts"))
         _validate_required_params_environment(self._get_param("environment"))
         _validate_required_params_project_name(self._get_param("name"))
@@ -125,7 +132,7 @@ class CodeBuildResponse(BaseResponse):
         if self._get_param("name") in self.codebuild_backend.codebuild_projects.keys():
             name = self._get_param("name")
             raise ResourceAlreadyExistsException(
-                f"Project already exists: arn:aws:codebuild:{self.region}:{self.current_account}:project/{name}"
+                f"Project already exists: arn:{get_partition(self.region)}:codebuild:{self.region}:{self.current_account}:project/{name}"
             )
 
         project_metadata = self.codebuild_backend.create_project(
@@ -151,7 +158,7 @@ class CodeBuildResponse(BaseResponse):
         ):
             name = self._get_param("projectName")
             raise ResourceNotFoundException(
-                f"Project cannot be found: arn:aws:codebuild:{self.region}:{self.current_account}:project/{name}"
+                f"Project cannot be found: arn:{get_partition(self.region)}:codebuild:{self.region}:{self.current_account}:project/{name}"
             )
 
         metadata = self.codebuild_backend.start_build(
diff --git a/moto/cognitoidp/models.py b/moto/cognitoidp/models.py
index 2b8917197..66a350a81 100644
--- a/moto/cognitoidp/models.py
+++ b/moto/cognitoidp/models.py
@@ -15,7 +15,7 @@ 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.utilities.paginator import paginate
-from moto.utilities.utils import md5_hash
+from moto.utilities.utils import get_partition, md5_hash
 
 from ..settings import get_cognito_idp_user_pool_id_strategy
 from .exceptions import (
@@ -396,7 +396,7 @@ class CognitoIdpUserPool(BaseModel):
             get_cognito_idp_user_pool_id_strategy(), region, name, extended_config
         )
         self.id = f"{self.region}_{user_pool_id}"[: self.MAX_ID_LENGTH]
-        self.arn = f"arn:aws:cognito-idp:{self.region}:{account_id}:userpool/{self.id}"
+        self.arn = f"arn:{get_partition(region)}:cognito-idp:{region}:{account_id}:userpool/{self.id}"
 
         self.name = name
         self.status = None
diff --git a/moto/core/base_backend.py b/moto/core/base_backend.py
index 0e29a6b7a..62564f03e 100644
--- a/moto/core/base_backend.py
+++ b/moto/core/base_backend.py
@@ -23,7 +23,7 @@ from moto.utilities.utils import get_partition
 
 from .model_instances import model_data
 from .responses import TYPE_RESPONSE
-from .utils import convert_regex_to_flask_path
+from .utils import ISO_REGION_DOMAINS, convert_regex_to_flask_path
 
 if TYPE_CHECKING:
     from moto.core.common_models import BaseModel
@@ -87,14 +87,7 @@ class BaseBackend:
             # This extension ensures support for the China & ISO regions
             alt_dns_suffixes = {"cn": "amazonaws.com.cn"}
             if enable_iso_regions():
-                alt_dns_suffixes.update(
-                    {
-                        "iso": "c2s.ic.gov",
-                        "isob": "sc2s.sgov.gov",
-                        "isoe": "cloud.adc-e.uk",
-                        "isof": "csp.hci.ic.gov",
-                    }
-                )
+                alt_dns_suffixes.update(ISO_REGION_DOMAINS)
 
             for url_path, handler in unformatted_paths.items():
                 url = url_path.format(url_base)
diff --git a/moto/core/responses.py b/moto/core/responses.py
index 3cd77f603..bc4c8ecc2 100644
--- a/moto/core/responses.py
+++ b/moto/core/responses.py
@@ -208,7 +208,7 @@ class ActionAuthenticatorMixin(object):
         self, bucket_name: Optional[str] = None, key_name: Optional[str] = None
     ) -> None:
         arn = f"{bucket_name or '*'}/{key_name}" if key_name else (bucket_name or "*")
-        resource = f"arn:aws:s3:::{arn}"
+        resource = f"arn:{get_partition(self.region)}:s3:::{arn}"  # type: ignore[attr-defined]
 
         from moto.iam.access_control import S3IAMRequest
 
diff --git a/moto/core/utils.py b/moto/core/utils.py
index 315069f74..ffea7a197 100644
--- a/moto/core/utils.py
+++ b/moto/core/utils.py
@@ -393,6 +393,15 @@ def gzip_decompress(body: bytes) -> bytes:
     return decompress(body)
 
 
+ISO_REGION_DOMAINS = {
+    "iso": "c2s.ic.gov",
+    "isob": "sc2s.sgov.gov",
+    "isoe": "cloud.adc-e.uk",
+    "isof": "csp.hci.ic.gov",
+}
+ALT_DOMAIN_SUFFIXES = list(ISO_REGION_DOMAINS.values()) + ["amazonaws.com.cn"]
+
+
 def get_equivalent_url_in_aws_domain(url: str) -> Tuple[ParseResult, bool]:
     """Parses a URL and converts non-standard AWS endpoint hostnames (from ISO
     regions or custom S3 endpoints) to the equivalent standard AWS domain.
@@ -406,14 +415,7 @@ def get_equivalent_url_in_aws_domain(url: str) -> Tuple[ParseResult, bool]:
 
     # https://github.com/getmoto/moto/pull/6412
     # Support ISO regions
-    iso_region_domains = [
-        "amazonaws.com.cn",
-        "c2s.ic.gov",
-        "sc2s.sgov.gov",
-        "cloud.adc-e.uk",
-        "csp.hci.ic.gov",
-    ]
-    for domain in iso_region_domains:
+    for domain in ALT_DOMAIN_SUFFIXES:
         if host.endswith(domain):
             host = host.replace(domain, "amazonaws.com")
 
diff --git a/moto/dynamodb/models/table.py b/moto/dynamodb/models/table.py
index 9a19a4dcb..e64f5eecb 100644
--- a/moto/dynamodb/models/table.py
+++ b/moto/dynamodb/models/table.py
@@ -211,9 +211,7 @@ class StreamShard(BaseModel):
         from moto.awslambda.utils import get_backend
 
         for arn, esm in self.table.lambda_event_source_mappings.items():
-            region = arn[
-                len("arn:aws:lambda:") : arn.index(":", len("arn:aws:lambda:"))
-            ]
+            region = arn.split(":")[3]
 
             result = get_backend(self.account_id, region).send_dynamodb_items(
                 arn, self.items, esm.event_source_arn
diff --git a/moto/ec2/models/managed_prefixes.py b/moto/ec2/models/managed_prefixes.py
index 8d50e3713..67d85de76 100644
--- a/moto/ec2/models/managed_prefixes.py
+++ b/moto/ec2/models/managed_prefixes.py
@@ -1,6 +1,6 @@
 from typing import Any, Dict, List, Optional
 
-from moto.utilities.utils import filter_resources
+from moto.utilities.utils import filter_resources, get_partition
 
 from ..utils import describe_tag_filter, random_managed_prefix_list_id
 from .core import TaggedEC2Resource
@@ -33,7 +33,9 @@ class ManagedPrefixList(TaggedEC2Resource):
         self.delete_counter = 1
 
     def arn(self, region: str, owner_id: str) -> str:
-        return f"arn:aws:ec2:{region}:{owner_id}:prefix-list/{self.id}"
+        return (
+            f"arn:{get_partition(region)}:ec2:{region}:{owner_id}:prefix-list/{self.id}"
+        )
 
     @property
     def owner_id(self) -> str:
diff --git a/moto/ec2/models/transit_gateway.py b/moto/ec2/models/transit_gateway.py
index e247d13ff..cdf34b5b8 100644
--- a/moto/ec2/models/transit_gateway.py
+++ b/moto/ec2/models/transit_gateway.py
@@ -52,7 +52,7 @@ class TransitGateway(TaggedEC2Resource, CloudFormationModel):
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:ec2:{self.ec2_backend.region_name}:{self.ec2_backend.account_id}:transit-gateway/{self.id}"
+        return f"arn:{self.ec2_backend.partition}:ec2:{self.ec2_backend.region_name}:{self.ec2_backend.account_id}:transit-gateway/{self.id}"
 
     @staticmethod
     def cloudformation_name_type() -> str:
diff --git a/moto/ec2/responses/launch_templates.py b/moto/ec2/responses/launch_templates.py
index f9df39a7a..648c36dbf 100644
--- a/moto/ec2/responses/launch_templates.py
+++ b/moto/ec2/responses/launch_templates.py
@@ -87,7 +87,7 @@ class LaunchTemplates(EC2BaseResponse):
             "launchTemplate",
             {
                 "createTime": version.create_time,
-                "createdBy": f"arn:aws:iam::{self.current_account}:root",
+                "createdBy": f"arn:{self.partition}:iam::{self.current_account}:root",
                 "defaultVersionNumber": template.default_version_number,
                 "latestVersionNumber": version.number,
                 "launchTemplateId": template.id,
@@ -120,7 +120,7 @@ class LaunchTemplates(EC2BaseResponse):
             "launchTemplateVersion",
             {
                 "createTime": version.create_time,
-                "createdBy": f"arn:aws:iam::{self.current_account}:root",
+                "createdBy": f"arn:{self.partition}:iam::{self.current_account}:root",
                 "defaultVersion": template.is_default(version),
                 "launchTemplateData": version.data,
                 "launchTemplateId": template.id,
@@ -214,7 +214,7 @@ class LaunchTemplates(EC2BaseResponse):
                 "item",
                 {
                     "createTime": version.create_time,
-                    "createdBy": f"arn:aws:iam::{self.current_account}:root",
+                    "createdBy": f"arn:{self.partition}:iam::{self.current_account}:root",
                     "defaultVersion": True,
                     "launchTemplateData": version.data,
                     "launchTemplateId": template.id,
@@ -251,7 +251,7 @@ class LaunchTemplates(EC2BaseResponse):
                 "item",
                 {
                     "createTime": template.create_time,
-                    "createdBy": f"arn:aws:iam::{self.current_account}:root",
+                    "createdBy": f"arn:{self.partition}:iam::{self.current_account}:root",
                     "defaultVersionNumber": template.default_version_number,
                     "latestVersionNumber": template.latest_version_number,
                     "launchTemplateId": template.id,
diff --git a/moto/ecs/models.py b/moto/ecs/models.py
index f25189531..703eb99ff 100644
--- a/moto/ecs/models.py
+++ b/moto/ecs/models.py
@@ -12,7 +12,7 @@ from moto.core.utils import pascal_to_camelcase, remap_nested_keys, unix_time
 from moto.ec2 import ec2_backends
 from moto.moto_api._internal import mock_random
 from moto.moto_api._internal.managed_state_model import ManagedState
-from moto.utilities.utils import get_partition
+from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition
 
 from ..ec2.utils import random_private_ip
 from .exceptions import (
@@ -1673,7 +1673,7 @@ class EC2ContainerServiceBackend(BaseBackend):
             if cluster_service_pair in self.services:
                 result.append(self.services[cluster_service_pair])
             else:
-                if name_or_arn.startswith("arn:aws:ecs"):
+                if re.match(ARN_PARTITION_REGEX + ":ecs", name_or_arn):
                     missing_arn = name_or_arn
                 else:
                     missing_arn = f"arn:{get_partition(self.region_name)}:ecs:{self.region_name}:{self.account_id}:service/{name}"
@@ -2040,9 +2040,12 @@ class EC2ContainerServiceBackend(BaseBackend):
     @staticmethod
     def _parse_resource_arn(resource_arn: str) -> Dict[str, str]:
         regexes = [
-            "^arn:aws:ecs:(?P<region>[^:]+):(?P<account_id>[^:]+):(?P<service>[^:]+)/(?P<cluster_id>[^:]+)/(?P<service_id>[^:]+)/ecs-svc/(?P<id>.*)$",
-            "^arn:aws:ecs:(?P<region>[^:]+):(?P<account_id>[^:]+):(?P<service>[^:]+)/(?P<cluster_id>[^:]+)/(?P<id>.*)$",
-            "^arn:aws:ecs:(?P<region>[^:]+):(?P<account_id>[^:]+):(?P<service>[^:]+)/(?P<id>.*)$",
+            ARN_PARTITION_REGEX
+            + ":ecs:(?P<region>[^:]+):(?P<account_id>[^:]+):(?P<service>[^:]+)/(?P<cluster_id>[^:]+)/(?P<service_id>[^:]+)/ecs-svc/(?P<id>.*)$",
+            ARN_PARTITION_REGEX
+            + ":ecs:(?P<region>[^:]+):(?P<account_id>[^:]+):(?P<service>[^:]+)/(?P<cluster_id>[^:]+)/(?P<id>.*)$",
+            ARN_PARTITION_REGEX
+            + ":ecs:(?P<region>[^:]+):(?P<account_id>[^:]+):(?P<service>[^:]+)/(?P<id>.*)$",
         ]
         for regex in regexes:
             match = re.match(regex, resource_arn)
diff --git a/moto/elasticbeanstalk/utils.py b/moto/elasticbeanstalk/utils.py
index e3e2e0f6b..bfeb7b861 100644
--- a/moto/elasticbeanstalk/utils.py
+++ b/moto/elasticbeanstalk/utils.py
@@ -1,10 +1,12 @@
+from moto.utilities.utils import get_partition
+
+
 def make_arn(
     region: str, account_id: str, resource_type: str, resource_path: str
 ) -> str:
-    arn_template = (
-        "arn:aws:elasticbeanstalk:{region}:{account_id}:{resource_type}/{resource_path}"
-    )
+    arn_template = "arn:{partition}:elasticbeanstalk:{region}:{account_id}:{resource_type}/{resource_path}"
     arn = arn_template.format(
+        partition=get_partition(region),
         region=region,
         account_id=account_id,
         resource_type=resource_type,
diff --git a/moto/elb/models.py b/moto/elb/models.py
index fcf5632bb..330bcc6b5 100644
--- a/moto/elb/models.py
+++ b/moto/elb/models.py
@@ -1,4 +1,5 @@
 import datetime
+import re
 from collections import OrderedDict
 from typing import Any, Dict, Iterable, List, Optional
 
@@ -7,6 +8,7 @@ from moto.core.common_models import BaseModel, CloudFormationModel
 from moto.ec2.exceptions import InvalidInstanceIdError
 from moto.ec2.models import ec2_backends
 from moto.moto_api._internal import mock_random
+from moto.utilities.utils import ARN_PARTITION_REGEX
 
 from .exceptions import (
     BadHealthCheckDefinition,
@@ -113,9 +115,8 @@ class FakeLoadBalancer(CloudFormationModel):
                     "ssl_certificate_id", port.get("SSLCertificateId")
                 ),
             )
-            if (
-                listener.ssl_certificate_id
-                and not listener.ssl_certificate_id.startswith("arn:aws:iam:")
+            if listener.ssl_certificate_id and not re.search(
+                f"{ARN_PARTITION_REGEX}:iam:", listener.ssl_certificate_id
             ):
                 elb_backend._register_certificate(
                     listener.ssl_certificate_id, self.dns_name
@@ -369,8 +370,8 @@ class ELBBackend(BaseBackend):
                             raise DuplicateListenerError(name, lb_port)
                         break
                 else:
-                    if ssl_certificate_id and not ssl_certificate_id.startswith(
-                        "arn:aws:iam::"
+                    if ssl_certificate_id and not re.search(
+                        f"{ARN_PARTITION_REGEX}:iam::", ssl_certificate_id
                     ):
                         self._register_certificate(
                             ssl_certificate_id, balancer.dns_name
diff --git a/moto/events/models.py b/moto/events/models.py
index 6cd308bcd..8bacd7bf8 100644
--- a/moto/events/models.py
+++ b/moto/events/models.py
@@ -1437,7 +1437,9 @@ class EventsBackend(BaseBackend):
                 "InvalidParameterValue", r"StatementId must match ^[a-zA-Z0-9-_]{1,64}$"
             )
 
-        principal_arn = {"AWS": f"arn:aws:iam::{principal}:root"}
+        principal_arn = {
+            "AWS": f"arn:{get_partition(self.region_name)}:iam::{principal}:root"
+        }
         stmt_condition = self._condition_param_to_stmt_condition(condition)
         event_bus.add_permission(statement_id, action, principal_arn, stmt_condition)
 
diff --git a/moto/events/notifications.py b/moto/events/notifications.py
index dd110ff7c..ba0dd3602 100644
--- a/moto/events/notifications.py
+++ b/moto/events/notifications.py
@@ -1,6 +1,8 @@
 import json
 from typing import Any
 
+from moto.utilities.utils import get_partition
+
 from .utils import EventMessageType
 
 _EVENT_S3_OBJECT_CREATED: EventMessageType = {
@@ -54,7 +56,9 @@ def _send_safe_notification(
                             applicable_targets.extend(rule.targets)
 
             for target in applicable_targets:
-                if target.get("Arn", "").startswith("arn:aws:lambda"):
+                if target.get("Arn", "").startswith(
+                    f"arn:{get_partition(region)}:lambda"
+                ):
                     _invoke_lambda(account_id, target.get("Arn"), event=event)  # type: ignore[arg-type]
 
 
diff --git a/moto/guardduty/models.py b/moto/guardduty/models.py
index 8e02b95fb..e3174b98a 100644
--- a/moto/guardduty/models.py
+++ b/moto/guardduty/models.py
@@ -4,6 +4,7 @@ from typing import Any, Dict, List, Optional
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.moto_api._internal import mock_random
+from moto.utilities.utils import get_partition
 
 from .exceptions import DetectorNotFoundException, FilterNotFoundException
 
@@ -30,6 +31,7 @@ class GuardDutyBackend(BaseBackend):
 
         detector = Detector(
             account_id=self.account_id,
+            region_name=self.region_name,
             created_at=datetime.now(),
             finding_publish_freq=finding_publishing_frequency,
             enabled=enable,
@@ -160,6 +162,7 @@ class Detector(BaseModel):
     def __init__(
         self,
         account_id: str,
+        region_name: str,
         created_at: datetime,
         finding_publish_freq: str,
         enabled: bool,
@@ -169,7 +172,7 @@ class Detector(BaseModel):
         self.id = mock_random.get_random_hex(length=32)
         self.created_at = created_at
         self.finding_publish_freq = finding_publish_freq
-        self.service_role = f"arn:aws:iam::{account_id}:role/aws-service-role/guardduty.amazonaws.com/AWSServiceRoleForAmazonGuardDuty"
+        self.service_role = f"arn:{get_partition(region_name)}:iam::{account_id}:role/aws-service-role/guardduty.amazonaws.com/AWSServiceRoleForAmazonGuardDuty"
         self.enabled = enabled
         self.updated_at = created_at
         self.datasources = datasources or {}
diff --git a/moto/iam/access_control.py b/moto/iam/access_control.py
index ec57fed82..c74c2bd49 100644
--- a/moto/iam/access_control.py
+++ b/moto/iam/access_control.py
@@ -95,7 +95,9 @@ class IAMUserAccessKey:
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:iam::{self.account_id}:user/{self._owner_user_name}"
+        return (
+            f"arn:{self.partition}:iam::{self.account_id}:user/{self._owner_user_name}"
+        )
 
     def create_credentials(self) -> Credentials:
         return Credentials(self._access_key_id, self._secret_access_key)
@@ -162,7 +164,7 @@ class AssumedRoleAccessKey:
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:sts::{self.account_id}:assumed-role/{self._owner_role_name}/{self._session_name}"
+        return f"arn:{self.partition}:sts::{self.account_id}:assumed-role/{self._owner_role_name}/{self._session_name}"
 
     def create_credentials(self) -> Credentials:
         return Credentials(
diff --git a/moto/iam/config.py b/moto/iam/config.py
index 093288d11..3d1b58759 100644
--- a/moto/iam/config.py
+++ b/moto/iam/config.py
@@ -1,4 +1,5 @@
 import json
+import re
 from typing import Any, Dict, List, Optional, Tuple
 
 import boto3
@@ -6,6 +7,7 @@ import boto3
 from moto.core.common_models import ConfigQueryModel
 from moto.core.exceptions import InvalidNextTokenException
 from moto.iam.models import IAMBackend, iam_backends
+from moto.utilities.utils import ARN_PARTITION_REGEX
 
 
 class RoleConfigQuery(ConfigQueryModel[IAMBackend]):
@@ -190,7 +192,9 @@ class PolicyConfigQuery(ConfigQueryModel[IAMBackend]):
         # custom configuration recorders, we'll just behave as default.
         policy_list = list(
             filter(
-                lambda policy: not policy.arn.startswith("arn:aws:iam::aws"),
+                lambda policy: not re.match(
+                    ARN_PARTITION_REGEX + ":iam::aws", policy.arn
+                ),
                 policy_list,
             )
         )
diff --git a/moto/iam/models.py b/moto/iam/models.py
index 39c4d7f7b..b052aa4aa 100644
--- a/moto/iam/models.py
+++ b/moto/iam/models.py
@@ -27,7 +27,12 @@ from moto.iam.policy_validation import (
 )
 from moto.moto_api._internal import mock_random as random
 from moto.settings import load_iam_aws_managed_policies
-from moto.utilities.utils import PARTITION_NAMES, get_partition, md5_hash
+from moto.utilities.utils import (
+    ARN_PARTITION_REGEX,
+    PARTITION_NAMES,
+    get_partition,
+    md5_hash,
+)
 
 from ..utilities.tagging_service import TaggingService
 from .aws_managed_policies import aws_managed_policies_data
@@ -116,8 +121,10 @@ class MFADevice:
 
 
 class VirtualMfaDevice:
-    def __init__(self, account_id: str, device_name: str):
-        self.serial_number = f"arn:aws:iam::{account_id}:mfa{device_name}"
+    def __init__(self, account_id: str, region_name: str, device_name: str):
+        self.serial_number = (
+            f"arn:{get_partition(region_name)}:iam::{account_id}:mfa{device_name}"
+        )
 
         random_base32_string = "".join(
             random.choice(string.ascii_uppercase + "234567") for _ in range(64)
@@ -209,21 +216,27 @@ class Policy(CloudFormationModel):
 
 class SAMLProvider(BaseModel):
     def __init__(
-        self, account_id: str, name: str, saml_metadata_document: Optional[str] = None
+        self,
+        account_id: str,
+        region_name: str,
+        name: str,
+        saml_metadata_document: Optional[str] = None,
     ):
         self.account_id = account_id
+        self.region_name = region_name
         self.name = name
         self.saml_metadata_document = saml_metadata_document
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:iam::{self.account_id}:saml-provider/{self.name}"
+        return f"arn:{get_partition(self.region_name)}:iam::{self.account_id}:saml-provider/{self.name}"
 
 
 class OpenIDConnectProvider(BaseModel):
     def __init__(
         self,
         account_id: str,
+        region_name: str,
         url: str,
         thumbprint_list: List[str],
         client_id_list: List[str],
@@ -233,6 +246,7 @@ class OpenIDConnectProvider(BaseModel):
         self._validate(url, thumbprint_list, client_id_list)
 
         self.account_id = account_id
+        self.region_name = region_name
         parsed_url = parse.urlparse(url)
         self.url = parsed_url.netloc + parsed_url.path
         self.thumbprint_list = thumbprint_list
@@ -242,7 +256,7 @@ class OpenIDConnectProvider(BaseModel):
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:iam::{self.account_id}:oidc-provider/{self.url}"
+        return f"arn:{get_partition(self.region_name)}:iam::{self.account_id}:oidc-provider/{self.url}"
 
     @property
     def created_iso_8601(self) -> str:
@@ -354,7 +368,9 @@ class ManagedPolicy(Policy, CloudFormationModel):
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:iam::{self.account_id}:policy{self.path}{self.name}"
+        return (
+            f"arn:{self.partition}:iam::{self.account_id}:policy{self.path}{self.name}"
+        )
 
     def to_config_dict(self) -> Dict[str, Any]:
         return {
@@ -362,7 +378,7 @@ class ManagedPolicy(Policy, CloudFormationModel):
             "configurationItemCaptureTime": str(self.create_date),
             "configurationItemStatus": "OK",
             "configurationStateId": str(int(unix_time())),
-            "arn": f"arn:aws:iam::{self.account_id}:policy/{self.name}",
+            "arn": f"arn:{self.partition}:iam::{self.account_id}:policy/{self.name}",
             "resourceType": "AWS::IAM::Policy",
             "resourceId": self.id,
             "resourceName": self.name,
@@ -373,7 +389,7 @@ class ManagedPolicy(Policy, CloudFormationModel):
             "configuration": {
                 "policyName": self.name,
                 "policyId": self.id,
-                "arn": f"arn:aws:iam::{self.account_id}:policy/{self.name}",
+                "arn": f"arn:{self.partition}:iam::{self.account_id}:policy/{self.name}",
                 "path": self.path,
                 "defaultVersionId": self.default_version_id,
                 "attachmentCount": self.attachment_count,
@@ -483,7 +499,7 @@ class AWSManagedPolicy(ManagedPolicy):
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:iam::aws:policy{self.path}{self.name}"
+        return f"arn:{self.partition}:iam::aws:policy{self.path}{self.name}"
 
 
 class InlinePolicy(CloudFormationModel):
@@ -764,8 +780,8 @@ class Role(CloudFormationModel):
     @property
     def arn(self) -> str:
         if self._linked_service:
-            return f"arn:aws:iam::{self.account_id}:role/aws-service-role/{self._linked_service}/{self.name}"
-        return f"arn:aws:iam::{self.account_id}:role{self.path}{self.name}"
+            return f"arn:{self.partition}:iam::{self.account_id}:role/aws-service-role/{self._linked_service}/{self.name}"
+        return f"arn:{self.partition}:iam::{self.account_id}:role{self.path}{self.name}"
 
     def to_config_dict(self) -> Dict[str, Any]:
         _managed_policies = []
@@ -797,7 +813,7 @@ class Role(CloudFormationModel):
             "configurationItemCaptureTime": str(self.create_date),
             "configurationItemStatus": "ResourceDiscovered",
             "configurationStateId": str(int(unix_time())),
-            "arn": f"arn:aws:iam::{self.account_id}:role/{self.name}",
+            "arn": f"arn:{self.partition}:iam::{self.account_id}:role/{self.name}",
             "resourceType": "AWS::IAM::Role",
             "resourceId": self.name,
             "resourceName": self.name,
@@ -811,7 +827,7 @@ class Role(CloudFormationModel):
                 "path": self.path,
                 "roleName": self.name,
                 "roleId": self.id,
-                "arn": f"arn:aws:iam::{self.account_id}:role/{self.name}",
+                "arn": f"arn:{self.partition}:iam::{self.account_id}:role/{self.name}",
                 "assumeRolePolicyDocument": parse.quote(
                     self.assume_role_policy_document
                 )
@@ -919,6 +935,7 @@ class InstanceProfile(CloudFormationModel):
     def __init__(
         self,
         account_id: str,
+        region_name: str,
         instance_profile_id: str,
         name: str,
         path: str,
@@ -927,6 +944,7 @@ class InstanceProfile(CloudFormationModel):
     ):
         self.id = instance_profile_id
         self.account_id = account_id
+        self.partition = get_partition(region_name)
         self.name = name
         self.path = path or "/"
         self.roles = roles if roles else []
@@ -983,7 +1001,7 @@ class InstanceProfile(CloudFormationModel):
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:iam::{self.account_id}:instance-profile{self.path}{self.name}"
+        return f"arn:{self.partition}:iam::{self.account_id}:instance-profile{self.path}{self.name}"
 
     @property
     def physical_resource_id(self) -> str:
@@ -1010,7 +1028,7 @@ class InstanceProfile(CloudFormationModel):
                     "path": role.path,
                     "roleName": role.name,
                     "roleId": role.id,
-                    "arn": f"arn:aws:iam::{self.account_id}:role/{role.name}",
+                    "arn": f"arn:{self.partition}:iam::{self.account_id}:role/{role.name}",
                     "createDate": str(role.create_date),
                     "assumeRolePolicyDocument": parse.quote(
                         role.assume_role_policy_document
@@ -1032,7 +1050,7 @@ class InstanceProfile(CloudFormationModel):
             "path": self.path,
             "instanceProfileName": self.name,
             "instanceProfileId": self.id,
-            "arn": f"arn:aws:iam::{self.account_id}:instance-profile/{role.name}",  # pylint: disable=W0631
+            "arn": f"arn:{self.partition}:iam::{self.account_id}:instance-profile/{role.name}",  # pylint: disable=W0631
             "createDate": str(self.create_date),
             "roles": roles,
         }
@@ -1042,6 +1060,7 @@ class Certificate(BaseModel):
     def __init__(
         self,
         account_id: str,
+        region_name: str,
         cert_name: str,
         cert_body: str,
         private_key: str,
@@ -1049,6 +1068,7 @@ class Certificate(BaseModel):
         path: Optional[str] = None,
     ):
         self.account_id = account_id
+        self.partition = get_partition(region_name)
         self.cert_name = cert_name
         if cert_body:
             cert_body = cert_body.rstrip()
@@ -1063,7 +1083,7 @@ class Certificate(BaseModel):
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:iam::{self.account_id}:server-certificate{self.path}{self.cert_name}"
+        return f"arn:{self.partition}:iam::{self.account_id}:server-certificate{self.path}{self.cert_name}"
 
 
 class SigningCertificate(BaseModel):
@@ -1235,8 +1255,9 @@ class SshPublicKey(BaseModel):
 
 
 class Group(BaseModel):
-    def __init__(self, account_id: str, name: str, path: str = "/"):
+    def __init__(self, account_id: str, region_name: str, name: str, path: str = "/"):
         self.account_id = account_id
+        self.partition = get_partition(region_name)
         self.name = name
         self.id = random_resource_id()
         self.path = path
@@ -1264,10 +1285,10 @@ class Group(BaseModel):
     @property
     def arn(self) -> str:
         if self.path == "/":
-            return f"arn:aws:iam::{self.account_id}:group/{self.name}"
+            return f"arn:{self.partition}:iam::{self.account_id}:group/{self.name}"
         else:
             # The path must by definition end and start with a forward slash. So we don't have to add more slashes to the ARN
-            return f"arn:aws:iam::{self.account_id}:group{self.path}{self.name}"
+            return f"arn:{self.partition}:iam::{self.account_id}:group{self.path}{self.name}"
 
     def get_policy(self, policy_name: str) -> Dict[str, str]:
         try:
@@ -1782,7 +1803,7 @@ class AccountSummary(BaseModel):
         customer_policies = [
             policy
             for policy in self._iam_backend.managed_policies
-            if not policy.startswith("arn:aws:iam::aws:policy")
+            if not re.match(ARN_PARTITION_REGEX + ":iam::aws:policy", policy)
         ]
         return len(customer_policies)
 
@@ -1834,7 +1855,9 @@ class IAMBackend(BaseBackend):
         self.account_aliases: List[str] = []
         self.saml_providers: Dict[str, SAMLProvider] = {}
         self.open_id_providers: Dict[str, OpenIDConnectProvider] = {}
-        self.policy_arn_regex = re.compile(r"^arn:aws:iam::(aws|[0-9]*):policy/.*$")
+        self.policy_arn_regex = re.compile(
+            ARN_PARTITION_REGEX + r":iam::(aws|[0-9]*):policy/.*$"
+        )
         self.virtual_mfa_devices: Dict[str, VirtualMfaDevice] = {}
         self.account_password_policy: Optional[AccountPasswordPolicy] = None
         self.account_summary = AccountSummary(self)
@@ -2390,7 +2413,13 @@ class IAMBackend(BaseBackend):
 
         roles = [self.get_role(role_name) for role_name in role_names]
         instance_profile = InstanceProfile(
-            self.account_id, instance_profile_id, name, path, roles, tags
+            account_id=self.account_id,
+            region_name=self.region_name,
+            instance_profile_id=instance_profile_id,
+            name=name,
+            path=path,
+            roles=roles,
+            tags=tags,
         )
         self.instance_profiles[name] = instance_profile
         return instance_profile
@@ -2466,7 +2495,13 @@ class IAMBackend(BaseBackend):
     ) -> Certificate:
         certificate_id = random_resource_id()
         cert = Certificate(
-            self.account_id, cert_name, cert_body, private_key, cert_chain, path
+            account_id=self.account_id,
+            region_name=self.region_name,
+            cert_name=cert_name,
+            cert_body=cert_body,
+            private_key=private_key,
+            cert_chain=cert_chain,
+            path=path,
         )
         self.certificates[certificate_id] = cert
         return cert
@@ -2504,7 +2539,7 @@ class IAMBackend(BaseBackend):
         if group_name in self.groups:
             raise IAMConflictException(f"Group {group_name} already exists")
 
-        group = Group(self.account_id, group_name, path)
+        group = Group(self.account_id, self.region_name, group_name, path)
         self.groups[group_name] = group
         return group
 
@@ -2959,7 +2994,11 @@ class IAMBackend(BaseBackend):
                 "Member must have length less than or equal to 512"
             )
 
-        device = VirtualMfaDevice(self.account_id, path + device_name)
+        device = VirtualMfaDevice(
+            self.account_id,
+            region_name=self.region_name,
+            device_name=path + device_name,
+        )
 
         if device.serial_number in self.virtual_mfa_devices:
             raise EntityAlreadyExists(
@@ -3075,7 +3114,12 @@ class IAMBackend(BaseBackend):
     def create_saml_provider(
         self, name: str, saml_metadata_document: str
     ) -> SAMLProvider:
-        saml_provider = SAMLProvider(self.account_id, name, saml_metadata_document)
+        saml_provider = SAMLProvider(
+            account_id=self.account_id,
+            region_name=self.region_name,
+            name=name,
+            saml_metadata_document=saml_metadata_document,
+        )
         self.saml_providers[name] = saml_provider
         return saml_provider
 
@@ -3120,7 +3164,12 @@ class IAMBackend(BaseBackend):
     ) -> OpenIDConnectProvider:
         clean_tags = self._tag_verification(tags)
         open_id_provider = OpenIDConnectProvider(
-            self.account_id, url, thumbprint_list, client_id_list, clean_tags
+            account_id=self.account_id,
+            region_name=self.region_name,
+            url=url,
+            thumbprint_list=thumbprint_list,
+            client_id_list=client_id_list,
+            tags=clean_tags,
         )
 
         if open_id_provider.arn in self.open_id_providers:
diff --git a/moto/iam/policy_validation.py b/moto/iam/policy_validation.py
index 5d47994a7..01e92f782 100644
--- a/moto/iam/policy_validation.py
+++ b/moto/iam/policy_validation.py
@@ -3,6 +3,7 @@ import re
 from typing import Any, Dict, List
 
 from moto.iam.exceptions import MalformedPolicyDocument
+from moto.utilities.utils import PARTITION_NAMES
 
 VALID_TOP_ELEMENTS = ["Version", "Id", "Statement", "Conditions"]
 
@@ -349,7 +350,10 @@ class BaseIAMPolicyValidator:
                 return
 
             resource_partitions = resource_partitions[2].partition(":")
-            if resource_partitions[0] not in ["aws", "*"]:
+            if (
+                resource_partitions[0] != "*"
+                and resource_partitions[0] not in PARTITION_NAMES
+            ):
                 remaining_resource_parts = resource_partitions[2].split(":")
 
                 arn1 = (
diff --git a/moto/inspector2/models.py b/moto/inspector2/models.py
index f2e797e11..814308005 100644
--- a/moto/inspector2/models.py
+++ b/moto/inspector2/models.py
@@ -6,6 +6,7 @@ from moto.core.common_models import BaseModel
 from moto.core.utils import unix_time
 from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 
 class FilterResource(BaseModel):
@@ -22,7 +23,7 @@ class FilterResource(BaseModel):
     ):
         filter_id = mock_random.get_random_hex(10)
         self.owner_id = account_id
-        self.arn = f"arn:aws:inspector2:{region}:{account_id}:owner/{self.owner_id}/filter/{filter_id}"
+        self.arn = f"arn:{get_partition(region)}:inspector2:{region}:{account_id}:owner/{self.owner_id}/filter/{filter_id}"
         self.name = name
         self.reason = reason
         self.action = action
diff --git a/moto/kms/models.py b/moto/kms/models.py
index 011233bc5..43d78940f 100644
--- a/moto/kms/models.py
+++ b/moto/kms/models.py
@@ -71,12 +71,12 @@ class Key(CloudFormationModel):
         self.id = generate_key_id(multi_region)
         self.creation_date = unix_time()
         self.account_id = account_id
+        self.region = region
         self.policy = policy or self.generate_default_policy()
         self.key_usage = key_usage
         self.key_state = "Enabled"
         self.description = description or ""
         self.enabled = True
-        self.region = region
         self.multi_region = multi_region
         self.key_rotation_status = False
         self.deletion_date: Optional[datetime] = None
@@ -144,7 +144,9 @@ class Key(CloudFormationModel):
                     {
                         "Sid": "Enable IAM User Permissions",
                         "Effect": "Allow",
-                        "Principal": {"AWS": f"arn:aws:iam::{self.account_id}:root"},
+                        "Principal": {
+                            "AWS": f"arn:{get_partition(self.region)}:iam::{self.account_id}:root"
+                        },
                         "Action": "kms:*",
                         "Resource": "*",
                     }
diff --git a/moto/medialive/models.py b/moto/medialive/models.py
index 434623b29..a5dd85a9c 100644
--- a/moto/medialive/models.py
+++ b/moto/medialive/models.py
@@ -4,6 +4,7 @@ from typing import Any, Dict, List, Optional
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.moto_api._internal import mock_random
+from moto.utilities.utils import get_partition
 
 
 class Input(BaseModel):
@@ -135,7 +136,7 @@ class MediaLiveBackend(BaseBackend):
         The RequestID and Reserved parameters are not yet implemented
         """
         channel_id = mock_random.uuid4().hex
-        arn = f"arn:aws:medialive:channel:{channel_id}"
+        arn = f"arn:{get_partition(self.region_name)}:medialive:channel:{channel_id}"
         channel = Channel(
             arn=arn,
             cdi_input_specification=cdi_input_specification,
@@ -231,7 +232,7 @@ class MediaLiveBackend(BaseBackend):
         The VPC and RequestId parameters are not yet implemented
         """
         input_id = mock_random.uuid4().hex
-        arn = f"arn:aws:medialive:input:{input_id}"
+        arn = f"arn:{get_partition(self.region_name)}:medialive:input:{input_id}"
         a_input = Input(
             arn=arn,
             input_id=input_id,
diff --git a/moto/mediapackage/models.py b/moto/mediapackage/models.py
index 5c228069e..ba05c1da4 100644
--- a/moto/mediapackage/models.py
+++ b/moto/mediapackage/models.py
@@ -3,6 +3,7 @@ from typing import Any, Dict, List
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
+from moto.utilities.utils import get_partition
 
 from .exceptions import ClientError
 
@@ -72,7 +73,7 @@ class MediaPackageBackend(BaseBackend):
     def create_channel(
         self, description: str, channel_id: str, tags: Dict[str, str]
     ) -> Channel:
-        arn = f"arn:aws:mediapackage:channel:{channel_id}"
+        arn = f"arn:{get_partition(self.region_name)}:mediapackage:channel:{channel_id}"
         channel = Channel(
             arn=arn,
             description=description,
@@ -117,7 +118,7 @@ class MediaPackageBackend(BaseBackend):
         time_delay_seconds: int,
         whitelist: List[str],
     ) -> OriginEndpoint:
-        arn = f"arn:aws:mediapackage:origin_endpoint:{endpoint_id}"
+        arn = f"arn:{get_partition(self.region_name)}:mediapackage:origin_endpoint:{endpoint_id}"
         url = f"https://origin-endpoint.mediapackage.{self.region_name}.amazonaws.com/{endpoint_id}"
         origin_endpoint = OriginEndpoint(
             arn=arn,
diff --git a/moto/mediastore/models.py b/moto/mediastore/models.py
index 4502d2d95..55473c7d2 100644
--- a/moto/mediastore/models.py
+++ b/moto/mediastore/models.py
@@ -4,6 +4,7 @@ from typing import Any, Dict, List, Optional
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
+from moto.utilities.utils import get_partition
 
 from .exceptions import (
     ContainerNotFoundException,
@@ -45,7 +46,7 @@ class MediaStoreBackend(BaseBackend):
         self._containers: Dict[str, Container] = OrderedDict()
 
     def create_container(self, name: str, tags: Dict[str, str]) -> Container:
-        arn = f"arn:aws:mediastore:container:{name}"
+        arn = f"arn:{get_partition(self.region_name)}:mediastore:container:{name}"
         container = Container(
             arn=arn,
             name=name,
diff --git a/moto/moto_proxy/proxy3.py b/moto/moto_proxy/proxy3.py
index eb8cd0bfb..6b20f8c55 100644
--- a/moto/moto_proxy/proxy3.py
+++ b/moto/moto_proxy/proxy3.py
@@ -16,6 +16,7 @@ from moto.backends import get_backend
 from moto.core import DEFAULT_ACCOUNT_ID
 from moto.core.base_backend import BackendDict
 from moto.core.exceptions import RESTError
+from moto.core.utils import get_equivalent_url_in_aws_domain
 from moto.moto_api._internal.models import moto_api_backend
 
 from . import debug, error, info, with_color
@@ -34,8 +35,13 @@ class MotoRequestHandler:
         if host == f"http://localhost:{self.port}":
             return "moto_api"
 
+        # Handle non-standard AWS endpoint hostnames from ISO regions or custom S3 endpoints.
+        parsed_url, _ = get_equivalent_url_in_aws_domain(host)
+        # Remove the querystring from the URL, as we'll never match on that
+        clean_url = f"{parsed_url.scheme}://{parsed_url.netloc}{parsed_url.path}"
+
         for backend, pattern in backend_url_patterns:
-            if pattern.match(host):
+            if pattern.match(clean_url):
                 return backend
 
     def get_handler_for_host(self, host: str, path: str) -> Any:
diff --git a/moto/neptune/models.py b/moto/neptune/models.py
index a50979c04..ee11a5686 100644
--- a/moto/neptune/models.py
+++ b/moto/neptune/models.py
@@ -17,6 +17,7 @@ class GlobalCluster(BaseModel):
     def __init__(
         self,
         account_id: str,
+        region_name: str,
         global_cluster_identifier: str,
         engine: Optional[str],
         engine_version: Optional[str],
@@ -25,9 +26,7 @@ class GlobalCluster(BaseModel):
     ):
         self.global_cluster_identifier = global_cluster_identifier
         self.global_cluster_resource_id = "cluster-" + random.get_random_hex(8)
-        self.global_cluster_arn = (
-            f"arn:aws:rds::{account_id}:global-cluster:{global_cluster_identifier}"
-        )
+        self.global_cluster_arn = f"arn:{get_partition(region_name)}:rds::{account_id}:global-cluster:{global_cluster_identifier}"
         self.engine = engine or "neptune"
         self.engine_version = engine_version or "1.2.0.0"
         self.storage_encrypted = (
@@ -298,6 +297,7 @@ class NeptuneBackend(BaseBackend):
     ) -> GlobalCluster:
         cluster = GlobalCluster(
             account_id=self.account_id,
+            region_name=self.region_name,
             global_cluster_identifier=global_cluster_identifier,
             engine=engine,
             engine_version=engine_version,
diff --git a/moto/pinpoint/models.py b/moto/pinpoint/models.py
index c83189b60..69053ce47 100644
--- a/moto/pinpoint/models.py
+++ b/moto/pinpoint/models.py
@@ -6,16 +6,15 @@ from moto.core.common_models import BaseModel
 from moto.core.utils import unix_time
 from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 from .exceptions import ApplicationNotFound, EventStreamNotFound
 
 
 class App(BaseModel):
-    def __init__(self, account_id: str, name: str):
+    def __init__(self, account_id: str, region_name: str, name: str):
         self.application_id = str(mock_random.uuid4()).replace("-", "")
-        self.arn = (
-            f"arn:aws:mobiletargeting:us-east-1:{account_id}:apps/{self.application_id}"
-        )
+        self.arn = f"arn:{get_partition(region_name)}:mobiletargeting:us-east-1:{account_id}:apps/{self.application_id}"
         self.name = name
         self.created = unix_time()
         self.settings = AppSettings()
@@ -95,7 +94,7 @@ class PinpointBackend(BaseBackend):
         self.tagger = TaggingService()
 
     def create_app(self, name: str, tags: Dict[str, str]) -> App:
-        app = App(self.account_id, name)
+        app = App(self.account_id, self.region_name, name)
         self.apps[app.application_id] = app
         tag_list = self.tagger.convert_dict_to_tags_input(tags)
         self.tagger.tag_resource(app.arn, tag_list)
diff --git a/moto/rds/models.py b/moto/rds/models.py
index d2776be07..82b475acf 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 get_partition, load_resource
+from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition, load_resource
 
 from .exceptions import (
     DBClusterNotFoundError,
@@ -73,6 +73,7 @@ class GlobalCluster(BaseModel):
     def __init__(
         self,
         account_id: str,
+        region_name: str,
         global_cluster_identifier: str,
         engine: str,
         engine_version: Optional[str],
@@ -81,9 +82,7 @@ class GlobalCluster(BaseModel):
     ):
         self.global_cluster_identifier = global_cluster_identifier
         self.global_cluster_resource_id = "cluster-" + random.get_random_hex(8)
-        self.global_cluster_arn = (
-            f"arn:aws:rds::{account_id}:global-cluster:{global_cluster_identifier}"
-        )
+        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 = (
@@ -1698,7 +1697,8 @@ class RDSBackend(BaseBackend):
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.arn_regex = re_compile(
-            r"^arn:aws:rds:.*:[0-9]*:(db|cluster|es|og|pg|ri|secgrp|snapshot|cluster-snapshot|subgrp|db-proxy):.*$"
+            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.global_clusters: Dict[str, GlobalCluster] = OrderedDict()
@@ -2892,7 +2892,7 @@ class RDSBackend(BaseBackend):
         source_cluster = None
         if source_db_cluster_identifier is not None:
             # validate our source cluster exists
-            if not source_db_cluster_identifier.startswith("arn:aws:rds"):
+            if not re.match(ARN_PARTITION_REGEX + ":rds", source_db_cluster_identifier):
                 raise InvalidParameterValue("Malformed db cluster arn dbci")
             source_cluster = self.describe_db_clusters(
                 cluster_identifier=source_db_cluster_identifier
@@ -2910,6 +2910,7 @@ class RDSBackend(BaseBackend):
             )
         global_cluster = GlobalCluster(
             account_id=self.account_id,
+            region_name=self.region_name,
             global_cluster_identifier=global_cluster_identifier,
             engine=engine,  # type: ignore
             engine_version=engine_version,
diff --git a/moto/resourcegroups/models.py b/moto/resourcegroups/models.py
index 1e05ce80d..d3f97c0c8 100644
--- a/moto/resourcegroups/models.py
+++ b/moto/resourcegroups/models.py
@@ -4,6 +4,7 @@ from typing import Any, Dict, List, Optional
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
+from moto.utilities.utils import get_partition
 
 from .exceptions import BadRequestException
 
@@ -12,6 +13,7 @@ class FakeResourceGroup(BaseModel):
     def __init__(
         self,
         account_id: str,
+        region_name: str,
         name: str,
         resource_query: Dict[str, str],
         description: Optional[str] = None,
@@ -30,7 +32,7 @@ class FakeResourceGroup(BaseModel):
         if self._validate_tags(value=tags):
             self._tags = tags
         self._raise_errors()
-        self.arn = f"arn:aws:resource-groups:us-west-1:{account_id}:{name}"
+        self.arn = f"arn:{get_partition(region_name)}:resource-groups:us-west-1:{account_id}:{name}"
         self.configuration = configuration
 
     @staticmethod
@@ -311,6 +313,7 @@ class ResourceGroupsBackend(BaseBackend):
         tags = tags or {}
         group = FakeResourceGroup(
             account_id=self.account_id,
+            region_name=self.region_name,
             name=name,
             resource_query=resource_query,
             description=description,
diff --git a/moto/resourcegroupstaggingapi/models.py b/moto/resourcegroupstaggingapi/models.py
index 2c9c131c7..3c2efb9d3 100644
--- a/moto/resourcegroupstaggingapi/models.py
+++ b/moto/resourcegroupstaggingapi/models.py
@@ -213,7 +213,7 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                     tags
                 ):  # Skip if no tags, or invalid filter
                     continue
-                yield {"ResourceARN": "arn:aws:s3:::" + bucket.name, "Tags": tags}
+                yield {"ResourceARN": bucket.arn, "Tags": tags}
 
         # CloudFormation
         if not resource_type_filters or "cloudformation:stack" in resource_type_filters:
@@ -267,7 +267,7 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                     # Skip if no tags, or invalid filter
                     continue
                 yield {
-                    "ResourceARN": f"arn:aws:ec2:{self.region_name}::image/{ami.id}",
+                    "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::image/{ami.id}",
                     "Tags": tags,
                 }
 
@@ -285,7 +285,7 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                         # Skip if no tags, or invalid filter
                         continue
                     yield {
-                        "ResourceARN": f"arn:aws:ec2:{self.region_name}::instance/{instance.id}",
+                        "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::instance/{instance.id}",
                         "Tags": tags,
                     }
 
@@ -302,7 +302,7 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                     # Skip if no tags, or invalid filter
                     continue
                 yield {
-                    "ResourceARN": f"arn:aws:ec2:{self.region_name}::network-interface/{eni.id}",
+                    "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::network-interface/{eni.id}",
                     "Tags": tags,
                 }
 
@@ -322,7 +322,7 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                         # Skip if no tags, or invalid filter
                         continue
                     yield {
-                        "ResourceARN": f"arn:aws:ec2:{self.region_name}::security-group/{sg.id}",
+                        "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::security-group/{sg.id}",
                         "Tags": tags,
                     }
 
@@ -339,7 +339,7 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                     # Skip if no tags, or invalid filter
                     continue
                 yield {
-                    "ResourceARN": f"arn:aws:ec2:{self.region_name}::snapshot/{snapshot.id}",
+                    "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::snapshot/{snapshot.id}",
                     "Tags": tags,
                 }
 
@@ -359,7 +359,7 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                 ):  # Skip if no tags, or invalid filter
                     continue
                 yield {
-                    "ResourceARN": f"arn:aws:ec2:{self.region_name}::volume/{volume.id}",
+                    "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}::volume/{volume.id}",
                     "Tags": tags,
                 }
 
@@ -601,7 +601,7 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                 ):  # Skip if no tags, or invalid filter
                     continue
                 yield {
-                    "ResourceARN": f"arn:aws:ec2:{self.region_name}:{self.account_id}:vpc/{vpc.id}",
+                    "ResourceARN": f"arn:{self.partition}:ec2:{self.region_name}:{self.account_id}:vpc/{vpc.id}",
                     "Tags": tags,
                 }
         # VPC Customer Gateway
@@ -940,18 +940,20 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
             "ErrorMessage": "Service not yet supported",
         }
         for arn in resource_arns:
-            if arn.startswith("arn:aws:rds:") or arn.startswith("arn:aws:snapshot:"):
+            if arn.startswith(
+                f"arn:{get_partition(self.region_name)}:rds:"
+            ) or arn.startswith(f"arn:{get_partition(self.region_name)}:snapshot:"):
                 self.rds_backend.add_tags_to_resource(
                     arn, TaggingService.convert_dict_to_tags_input(tags)
                 )
-            elif arn.startswith("arn:aws:workspaces:"):
+            elif arn.startswith(f"arn:{get_partition(self.region_name)}:workspaces:"):
                 resource_id = arn.split("/")[-1]
                 self.workspaces_backend.create_tags(  # type: ignore[union-attr]
                     resource_id, TaggingService.convert_dict_to_tags_input(tags)
                 )
-            elif arn.startswith("arn:aws:logs:"):
+            elif arn.startswith(f"arn:{get_partition(self.region_name)}:logs:"):
                 self.logs_backend.tag_resource(arn, tags)
-            elif arn.startswith("arn:aws:dynamodb"):
+            elif arn.startswith(f"arn:{get_partition(self.region_name)}:dynamodb"):
                 self.dynamodb_backend.tag_resource(
                     arn, TaggingService.convert_dict_to_tags_input(tags)
                 )
diff --git a/moto/route53/models.py b/moto/route53/models.py
index 89a187471..ec2b8b816 100644
--- a/moto/route53/models.py
+++ b/moto/route53/models.py
@@ -457,13 +457,14 @@ class FakeZone(CloudFormationModel):
 
 
 class RecordSetGroup(CloudFormationModel):
-    def __init__(self, hosted_zone_id: str, record_sets: List[str]):
+    def __init__(self, region_name: str, hosted_zone_id: str, record_sets: List[str]):
+        self.region_name = region_name
         self.hosted_zone_id = hosted_zone_id
         self.record_sets = record_sets
 
     @property
     def physical_resource_id(self) -> str:
-        return f"arn:aws:route53:::hostedzone/{self.hosted_zone_id}"
+        return f"arn:{get_partition(self.region_name)}:route53:::hostedzone/{self.hosted_zone_id}"
 
     @staticmethod
     def cloudformation_name_type() -> str:
@@ -494,7 +495,7 @@ class RecordSetGroup(CloudFormationModel):
         for record_set in record_sets:
             hosted_zone.add_rrset(record_set)
 
-        return RecordSetGroup(hosted_zone.id, record_sets)
+        return RecordSetGroup(region_name, hosted_zone.id, record_sets)
 
 
 class QueryLoggingConfig(BaseModel):
diff --git a/moto/route53resolver/models.py b/moto/route53resolver/models.py
index 3f02147a9..a4e41fe45 100644
--- a/moto/route53resolver/models.py
+++ b/moto/route53resolver/models.py
@@ -24,6 +24,7 @@ from moto.route53resolver.utils import PAGINATION_MODEL
 from moto.route53resolver.validations import validate_args
 from moto.utilities.paginator import paginate
 from moto.utilities.tagging_service import TaggingService
+from moto.utilities.utils import get_partition
 
 CAMEL_TO_SNAKE_PATTERN = re.compile(r"(?<!^)(?=[A-Z])")
 
@@ -129,8 +130,7 @@ class ResolverRule(BaseModel):  # pylint: disable=too-many-instance-attributes
 
     @property
     def arn(self) -> str:
-        """Return ARN for this resolver rule."""
-        return f"arn:aws:route53resolver:{self.region}:{self.account_id}:resolver-rule/{self.id}"
+        return f"arn:{get_partition(self.region)}:route53resolver:{self.region}:{self.account_id}:resolver-rule/{self.id}"
 
     def description(self) -> Dict[str, Any]:
         """Return a dictionary of relevant info for this resolver rule."""
@@ -213,8 +213,7 @@ class ResolverEndpoint(BaseModel):  # pylint: disable=too-many-instance-attribut
 
     @property
     def arn(self) -> str:
-        """Return ARN for this resolver endpoint."""
-        return f"arn:aws:route53resolver:{self.region}:{self.account_id}:resolver-endpoint/{self.id}"
+        return f"arn:{get_partition(self.region)}:route53resolver:{self.region}:{self.account_id}:resolver-endpoint/{self.id}"
 
     def _vpc_id_from_subnet(self) -> str:
         """Return VPC Id associated with the subnet.
diff --git a/moto/s3/models.py b/moto/s3/models.py
index b238af776..5ee42576e 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -192,7 +192,7 @@ class FakeKey(BaseModel, ManagedState):
     @property
     def arn(self) -> str:
         # S3 Objects don't have an ARN, but we do need something unique when creating tags against this resource
-        return f"arn:aws:s3:::{self.bucket_name}/{self.name}/{self.version_id}"
+        return f"arn:{self.partition}:s3:::{self.bucket_name}/{self.name}/{self.version_id}"
 
     @value.setter  # type: ignore
     def value(self, new_value: bytes) -> None:
@@ -1041,6 +1041,7 @@ class FakeBucket(CloudFormationModel):
         self.name = name
         self.account_id = account_id
         self.region_name = region_name
+        self.partition = get_partition(region_name)
         self.keys = _VersionedKeyStore()
         self.multiparts = MultipartDict()
         self.versioning_status: Optional[str] = None
@@ -1298,9 +1299,11 @@ class FakeBucket(CloudFormationModel):
                     continue
                 if (
                     stmt.get("Resource")
-                    != f"arn:aws:s3:::{target_bucket.name}/{target_prefix if target_prefix else ''}*"
-                    and stmt.get("Resource") != f"arn:aws:s3:::{target_bucket.name}/*"
-                    and stmt.get("Resource") != f"arn:aws:s3:::{target_bucket.name}"
+                    != f"arn:{target_bucket.partition}:s3:::{target_bucket.name}/{target_prefix if target_prefix else ''}*"
+                    and stmt.get("Resource")
+                    != f"arn:{target_bucket.partition}:s3:::{target_bucket.name}/*"
+                    and stmt.get("Resource")
+                    != f"arn:{target_bucket.partition}:s3:::{target_bucket.name}"
                 ):
                     continue
                 return True
@@ -1424,7 +1427,7 @@ class FakeBucket(CloudFormationModel):
 
     @property
     def arn(self) -> str:
-        return f"arn:aws:s3:::{self.name}"
+        return f"arn:{self.partition}:s3:::{self.name}"
 
     @property
     def domain_name(self) -> str:
@@ -1825,7 +1828,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
             source="aws.s3",
             event_name="CreateBucket",
             region=region_name,
-            resources=[f"arn:aws:s3:::{bucket_name}"],
+            resources=[f"arn:{new_bucket.partition}:s3:::{bucket_name}"],
             detail=notification_detail,
         )
 
@@ -1842,9 +1845,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
             if not s3_allow_crossdomain_access():
                 raise AccessDeniedByLock
             account_id = s3_backends.bucket_accounts[bucket_name]
-            return s3_backends[account_id][get_partition(self.region_name)].get_bucket(
-                bucket_name
-            )
+            return s3_backends[account_id][self.partition].get_bucket(bucket_name)
 
         raise MissingBucket(bucket=bucket_name)
 
diff --git a/moto/s3/notifications.py b/moto/s3/notifications.py
index bc9208695..9f935a8f2 100644
--- a/moto/s3/notifications.py
+++ b/moto/s3/notifications.py
@@ -2,10 +2,13 @@ import copy
 import json
 from datetime import datetime
 from enum import Enum
-from typing import Any, Dict, List
+from typing import TYPE_CHECKING, Any, Dict, List
 
 from moto.core.utils import unix_time
 
+if TYPE_CHECKING:
+    from moto.s3.models import FakeBucket
+
 _EVENT_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"
 
 
@@ -64,7 +67,7 @@ class S3NotificationEvent(str, Enum):
 
 
 def _get_s3_event(
-    event_name: str, bucket: Any, key: Any, notification_id: str
+    event_name: str, bucket: "FakeBucket", key: Any, notification_id: str
 ) -> Dict[str, List[Dict[str, Any]]]:
     etag = key.etag.replace('"', "")
     # s3:ObjectCreated:Put --> ObjectCreated:Put
@@ -83,7 +86,7 @@ def _get_s3_event(
                     "configurationId": notification_id,
                     "bucket": {
                         "name": bucket.name,
-                        "arn": f"arn:aws:s3:::{bucket.name}",
+                        "arn": bucket.arn,
                     },
                     "object": {"key": key.name, "size": key.size, "eTag": etag},
                 },
@@ -165,7 +168,7 @@ def _send_sns_message(
 
 def _send_event_bridge_message(
     account_id: str,
-    bucket: Any,
+    bucket: "FakeBucket",
     event_name: str,
     key: Any,
 ) -> None:
@@ -179,7 +182,7 @@ def _send_event_bridge_message(
         event["account"] = account_id
         event["time"] = unix_time()
         event["region"] = bucket.region_name
-        event["resources"] = [f"arn:aws:s3:::{bucket.name}"]
+        event["resources"] = [bucket.arn]
         event["detail"] = {
             "version": "0",
             "bucket": {"name": bucket.name},
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index 66c758899..dd9e9c91c 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -11,6 +11,7 @@ from moto import settings
 from moto.core.common_types import TYPE_RESPONSE
 from moto.core.responses import BaseResponse
 from moto.core.utils import (
+    ALT_DOMAIN_SUFFIXES,
     extract_region_from_aws_authorization,
     path_url,
     str_to_rfc_1123_datetime,
@@ -21,7 +22,7 @@ from moto.s3bucket_path.utils import (
 from moto.s3bucket_path.utils import (
     parse_key_name as bucketpath_parse_key_name,
 )
-from moto.utilities.utils import get_partition
+from moto.utilities.utils import PARTITION_NAMES, get_partition
 
 from .exceptions import (
     AccessForbidden,
@@ -237,8 +238,13 @@ class S3Response(BaseResponse):
             if match:
                 return False
 
-        path_based = host == "s3.amazonaws.com" or re.match(
-            r"s3[\.\-]([^.]*)\.amazonaws\.com", host
+        path_based = (
+            host == "s3.amazonaws.com"
+            or re.match(r"s3[\.\-]([^.]*)\.amazonaws\.com", host)
+            or any(
+                re.match(r"s3[\.\-]([^.]*)\." + suffix, host)
+                for suffix in ALT_DOMAIN_SUFFIXES
+            )
         )
         return not path_based
 
@@ -1288,7 +1294,7 @@ class S3Response(BaseResponse):
         except S3ClientError:
             key = bucket = None
         if key:
-            resource = f"arn:aws:s3:::{bucket_name}/{key_name}"
+            resource = f"arn:{bucket.partition}:s3:::{bucket_name}/{key_name}"  # type: ignore[union-attr]
 
             # Authorization Workflow
             # https://docs.aws.amazon.com/AmazonS3/latest/userguide/access-control-auth-workflow-object-operation.html
@@ -2125,7 +2131,10 @@ class S3Response(BaseResponse):
                     ] = [the_notification]
 
                 for n in the_notification:
-                    if not n[name].startswith(f"arn:aws:{arn_string}:"):
+                    if not any(
+                        n[name].startswith(f"arn:{p}:{arn_string}:")
+                        for p in PARTITION_NAMES
+                    ):
                         raise InvalidNotificationARN()
 
                     # 2nd, verify that the Events list is correct:
diff --git a/moto/s3control/models.py b/moto/s3control/models.py
index 76a05a26b..381eb745e 100644
--- a/moto/s3control/models.py
+++ b/moto/s3control/models.py
@@ -11,7 +11,7 @@ from moto.s3.exceptions import (
     WrongPublicAccessBlockAccountIdError,
 )
 from moto.s3.models import PublicAccessBlock
-from moto.utilities.utils import PARTITION_NAMES
+from moto.utilities.utils import PARTITION_NAMES, get_partition
 
 from .exceptions import AccessPointNotFound, AccessPointPolicyNotFound
 
@@ -20,6 +20,7 @@ class AccessPoint(BaseModel):
     def __init__(
         self,
         account_id: str,
+        region_name: str,
         name: str,
         bucket: str,
         vpc_configuration: Dict[str, Any],
@@ -29,7 +30,7 @@ class AccessPoint(BaseModel):
         self.alias = f"{name}-{mock_random.get_random_hex(34)}-s3alias"
         self.bucket = bucket
         self.created = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")
-        self.arn = f"arn:aws:s3:us-east-1:{account_id}:accesspoint/{name}"
+        self.arn = f"arn:{get_partition(region_name)}:s3:us-east-1:{account_id}:accesspoint/{name}"
         self.policy: Optional[str] = None
         self.network_origin = "VPC" if vpc_configuration else "Internet"
         self.vpc_id = (vpc_configuration or {}).get("VpcId")
@@ -101,10 +102,11 @@ class S3ControlBackend(BaseBackend):
     ) -> AccessPoint:
         access_point = AccessPoint(
             account_id,
-            name,
-            bucket,
-            vpc_configuration,
-            public_access_block_configuration,
+            region_name=self.region_name,
+            name=name,
+            bucket=bucket,
+            vpc_configuration=vpc_configuration,
+            public_access_block_configuration=public_access_block_configuration,
         )
         self.access_points[account_id][name] = access_point
         return access_point
diff --git a/moto/sagemaker/models.py b/moto/sagemaker/models.py
index ee8425a30..7da3f8d91 100644
--- a/moto/sagemaker/models.py
+++ b/moto/sagemaker/models.py
@@ -1,6 +1,7 @@
 import json
 import os
 import random
+import re
 import string
 from datetime import datetime
 from typing import Any, Dict, Iterable, List, Optional, Union, cast
@@ -11,7 +12,7 @@ from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel, CloudFormationModel
 from moto.sagemaker import validators
 from moto.utilities.paginator import paginate
-from moto.utilities.utils import get_partition
+from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition
 
 from .exceptions import (
     AWSValidationException,
@@ -2254,7 +2255,7 @@ class SageMakerModelBackend(BaseBackend):
             self.trials[trial_name].trial_components.extend([trial_component_name])
         else:
             raise ResourceNotFound(
-                message=f"Trial 'arn:aws:sagemaker:{self.region_name}:{self.account_id}:experiment-trial/{trial_name}' does not exist."
+                message=f"Trial 'arn:{get_partition(self.region_name)}:sagemaker:{self.region_name}:{self.account_id}:experiment-trial/{trial_name}' does not exist."
             )
 
         if trial_component_name in self.trial_components.keys():
@@ -3450,7 +3451,7 @@ class SageMakerModelBackend(BaseBackend):
             )
         if model_package_group_name is not None:
             model_package_type = "Versioned"
-            if model_package_group_name.startswith("arn:aws"):
+            if re.match(ARN_PARTITION_REGEX, model_package_group_name):
                 model_package_group_name = model_package_group_name.split("/")[-1]
         model_package_summary_list = list(
             filter(
diff --git a/moto/secretsmanager/utils.py b/moto/secretsmanager/utils.py
index d58c9d3ca..e5435a6c9 100644
--- a/moto/secretsmanager/utils.py
+++ b/moto/secretsmanager/utils.py
@@ -2,7 +2,7 @@ import re
 import string
 
 from moto.moto_api._internal import mock_random as random
-from moto.utilities.utils import get_partition
+from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition
 
 
 def random_password(
@@ -76,7 +76,7 @@ def get_secret_name_from_partial_arn(partial_arn: str) -> str:
     # This method only deals with partial ARN's, and will return the name: testsecret
     #
     # If you were to pass in  full url, this method will return 'testsecret-xxxxxx' - which has no meaning on it's own
-    if partial_arn.startswith("arn:aws:secretsmanager:"):
+    if re.match(ARN_PARTITION_REGEX + ":secretsmanager:", partial_arn):
         # split the arn by colon
         # then get the last value which is the name appended with a random string
         return partial_arn.split(":")[-1]
diff --git a/moto/sns/models.py b/moto/sns/models.py
index 4ca029d35..7b85f2373 100644
--- a/moto/sns/models.py
+++ b/moto/sns/models.py
@@ -959,6 +959,7 @@ class SNSBackend(BaseBackend):
 
     def add_permission(
         self,
+        region_name: str,
         topic_arn: str,
         label: str,
         aws_account_ids: List[str],
@@ -982,7 +983,8 @@ class SNSBackend(BaseBackend):
             raise SNSInvalidParameter("Policy statement action out of service scope!")
 
         principals = [
-            f"arn:aws:iam::{account_id}:root" for account_id in aws_account_ids
+            f"arn:{get_partition(region_name)}:iam::{account_id}:root"
+            for account_id in aws_account_ids
         ]
         actions = [f"SNS:{action_name}" for action_name in action_names]
 
diff --git a/moto/sns/responses.py b/moto/sns/responses.py
index 8860c0dfc..b293aff50 100644
--- a/moto/sns/responses.py
+++ b/moto/sns/responses.py
@@ -733,7 +733,13 @@ class SNSResponse(BaseResponse):
         aws_account_ids = self._get_multi_param("AWSAccountId.member.")
         action_names = self._get_multi_param("ActionName.member.")
 
-        self.backend.add_permission(topic_arn, label, aws_account_ids, action_names)
+        self.backend.add_permission(
+            region_name=self.region,
+            topic_arn=topic_arn,
+            label=label,
+            aws_account_ids=aws_account_ids,
+            action_names=action_names,
+        )
 
         template = self.response_template(ADD_PERMISSION_TEMPLATE)
         return template.render()
diff --git a/moto/sqs/models.py b/moto/sqs/models.py
index 2d4deefcf..35f6f1764 100644
--- a/moto/sqs/models.py
+++ b/moto/sqs/models.py
@@ -1155,7 +1155,12 @@ class SQSBackend(BaseBackend):
         return queues
 
     def add_permission(
-        self, queue_name: str, actions: List[str], account_ids: List[str], label: str
+        self,
+        region_name: str,
+        queue_name: str,
+        actions: List[str],
+        account_ids: List[str],
+        label: str,
     ) -> None:
         queue = self.get_queue(queue_name)
 
@@ -1195,7 +1200,10 @@ class SQSBackend(BaseBackend):
                 f"Value {label} for parameter Label is invalid. Reason: Already exists."
             )
 
-        principals = [f"arn:aws:iam::{account_id}:root" for account_id in account_ids]
+        principals = [
+            f"arn:{get_partition(region_name)}:iam::{account_id}:root"
+            for account_id in account_ids
+        ]
         actions = [f"SQS:{action}" for action in actions]
 
         statement = {
diff --git a/moto/sqs/responses.py b/moto/sqs/responses.py
index 613a4594c..3c84d2710 100644
--- a/moto/sqs/responses.py
+++ b/moto/sqs/responses.py
@@ -684,7 +684,13 @@ class SQSResponse(BaseResponse):
         )
         label = self._get_param("Label")
 
-        self.sqs_backend.add_permission(queue_name, actions, account_ids, label)
+        self.sqs_backend.add_permission(
+            region_name=self.region,
+            queue_name=queue_name,
+            actions=actions,
+            account_ids=account_ids,
+            label=label,
+        )
 
         return self._empty_response(ADD_PERMISSION_RESPONSE)
 
diff --git a/moto/ssoadmin/models.py b/moto/ssoadmin/models.py
index ecf148910..7900bdb03 100644
--- a/moto/ssoadmin/models.py
+++ b/moto/ssoadmin/models.py
@@ -215,7 +215,8 @@ class SSOAdminBackend(BaseBackend):
         managed_policy = self.aws_managed_policies.get(policy_name, None)
         if managed_policy is not None:
             path = managed_policy.get("path", "/")
-            expected_arn = f"arn:aws:iam::aws:policy{path}{policy_name}"
+            expected_arn = f"arn:{self.partition}:iam::aws:policy{path}{policy_name}"
+
             if managed_policy_arn == expected_arn:
                 return ManagedPolicy(managed_policy_arn, policy_name)
         raise ResourceNotFoundException(
diff --git a/moto/stepfunctions/models.py b/moto/stepfunctions/models.py
index 5c4507142..d2ea33a56 100644
--- a/moto/stepfunctions/models.py
+++ b/moto/stepfunctions/models.py
@@ -11,7 +11,7 @@ from moto.core.common_models import CloudFormationModel
 from moto.core.utils import iso_8601_datetime_with_milliseconds
 from moto.moto_api._internal import mock_random
 from moto.utilities.paginator import paginate
-from moto.utilities.utils import get_partition
+from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition
 
 from .exceptions import (
     ExecutionAlreadyExists,
@@ -251,9 +251,13 @@ class Execution:
         state_machine_arn: str,
         execution_input: str,
     ):
-        execution_arn = "arn:aws:states:{}:{}:execution:{}:{}"
+        execution_arn = "arn:{}:states:{}:{}:execution:{}:{}"
         execution_arn = execution_arn.format(
-            region_name, account_id, state_machine_name, execution_name
+            get_partition(region_name),
+            region_name,
+            account_id,
+            state_machine_name,
+            execution_name,
         )
         self.execution_arn = execution_arn
         self.name = execution_name
@@ -495,13 +499,15 @@ class StepFunctionBackend(BaseBackend):
         "\u009f",
     ]
     accepted_role_arn_format = re.compile(
-        r"arn:(aws|aws-cn|aws-us-gov|aws-iso|aws-iso-b):iam::(?P<account_id>[0-9]{12}):role/.+"
+        ARN_PARTITION_REGEX + r":iam::(?P<account_id>[0-9]{12}):role/.+"
     )
     accepted_mchn_arn_format = re.compile(
-        r"arn:(aws|aws-cn|aws-us-gov|aws-iso|aws-iso-b):states:[-0-9a-zA-Z]+:(?P<account_id>[0-9]{12}):stateMachine:.+"
+        ARN_PARTITION_REGEX
+        + r":states:[-0-9a-zA-Z]+:(?P<account_id>[0-9]{12}):stateMachine:.+"
     )
     accepted_exec_arn_format = re.compile(
-        r"arn:(aws|aws-cn|aws-us-gov|aws-iso|aws-iso-b):states:[-0-9a-zA-Z]+:(?P<account_id>[0-9]{12}):execution:.+"
+        ARN_PARTITION_REGEX
+        + r":states:[-0-9a-zA-Z]+:(?P<account_id>[0-9]{12}):execution:.+"
     )
 
     def __init__(self, region_name: str, account_id: str):
diff --git a/moto/sts/models.py b/moto/sts/models.py
index 02f844e48..68db185d0 100644
--- a/moto/sts/models.py
+++ b/moto/sts/models.py
@@ -14,7 +14,7 @@ from moto.sts.utils import (
     random_assumed_role_id,
     random_session_token,
 )
-from moto.utilities.utils import PARTITION_NAMES, get_partition
+from moto.utilities.utils import ARN_PARTITION_REGEX, PARTITION_NAMES, get_partition
 
 
 class Token(BaseModel):
@@ -192,9 +192,9 @@ class STSBackend(BaseBackend):
         return user_id, arn, self.account_id
 
     def _create_access_key(self, role: str) -> Tuple[str, AccessKey]:
-        account_id_match = re.search(r"arn:aws:iam::([0-9]+).+", role)
+        account_id_match = re.search(ARN_PARTITION_REGEX + r":iam::([0-9]+).+", role)
         if account_id_match:
-            account_id = account_id_match.group(1)
+            account_id = account_id_match.group(2)
         else:
             account_id = self.account_id
         iam_backend = iam_backends[account_id][self.partition]
diff --git a/moto/utilities/utils.py b/moto/utilities/utils.py
index 4804269e7..77fae680e 100644
--- a/moto/utilities/utils.py
+++ b/moto/utilities/utils.py
@@ -3,6 +3,7 @@ import json
 import pkgutil
 from typing import Any, Dict, Iterator, List, MutableMapping, Optional, Tuple, TypeVar
 
+DEFAULT_PARTITION = "aws"
 REGION_PREFIX_TO_PARTITION = {
     # (region prefix, aws partition)
     "cn-": "aws-cn",
@@ -10,16 +11,19 @@ REGION_PREFIX_TO_PARTITION = {
     "us-iso-": "aws-iso",
     "us-isob-": "aws-iso-b",
 }
-PARTITION_NAMES = list(REGION_PREFIX_TO_PARTITION.values()) + ["aws"]
+PARTITION_NAMES = list(REGION_PREFIX_TO_PARTITION.values()) + [DEFAULT_PARTITION]
+ARN_PARTITION_REGEX = r"^arn:(" + "|".join(sorted(PARTITION_NAMES)) + ")"
 
 
 def get_partition(region: str) -> str:
+    if not region:
+        return DEFAULT_PARTITION
     if region in PARTITION_NAMES:
         return region
     for prefix in REGION_PREFIX_TO_PARTITION:
         if region.startswith(prefix):
             return REGION_PREFIX_TO_PARTITION[prefix]
-    return "aws"
+    return DEFAULT_PARTITION
 
 
 def str2bool(v: Any) -> Optional[bool]:
diff --git a/moto/wafv2/models.py b/moto/wafv2/models.py
index 5d960ffe0..173cb1658 100644
--- a/moto/wafv2/models.py
+++ b/moto/wafv2/models.py
@@ -7,7 +7,7 @@ from moto.core.common_models import BaseModel
 from moto.core.utils import iso_8601_datetime_with_milliseconds
 from moto.moto_api._internal import mock_random
 from moto.utilities.tagging_service import TaggingService
-from moto.utilities.utils import PARTITION_NAMES
+from moto.utilities.utils import ARN_PARTITION_REGEX, PARTITION_NAMES
 
 from .exceptions import (
     WAFNonexistentItemException,
@@ -23,7 +23,8 @@ if TYPE_CHECKING:
 US_EAST_1_REGION = "us-east-1"
 GLOBAL_REGION = "global"
 APIGATEWAY_REGEX = (
-    r"arn:aws:apigateway:[a-zA-Z0-9-]+::/restapis/[a-zA-Z0-9]+/stages/[a-zA-Z0-9]+"
+    ARN_PARTITION_REGEX
+    + r":apigateway:[a-zA-Z0-9-]+::/restapis/[a-zA-Z0-9]+/stages/[a-zA-Z0-9]+"
 )
 
 
diff --git a/moto/wafv2/utils.py b/moto/wafv2/utils.py
index 071a41967..9bb7b4753 100644
--- a/moto/wafv2/utils.py
+++ b/moto/wafv2/utils.py
@@ -1,4 +1,4 @@
-from moto.utilities.utils import PARTITION_NAMES
+from moto.utilities.utils import PARTITION_NAMES, get_partition
 
 
 def make_arn_for_wacl(
@@ -30,5 +30,8 @@ def make_arn(
 
     if region_name in PARTITION_NAMES:
         region_name = "global"
+    partition = (
+        region_name if region_name in PARTITION_NAMES else get_partition(region_name)
+    )
 
-    return f"arn:aws:wafv2:{region_name}:{account_id}:{scope}/{resource}/{name}/{_id}"
+    return f"arn:{partition}:wafv2:{region_name}:{account_id}:{scope}/{resource}/{name}/{_id}"
diff --git a/moto/workspaces/models.py b/moto/workspaces/models.py
index 86cdb5338..87c7b09eb 100644
--- a/moto/workspaces/models.py
+++ b/moto/workspaces/models.py
@@ -11,6 +11,7 @@ from moto.ds import ds_backends
 from moto.ds.models import Directory
 from moto.ec2 import ec2_backends
 from moto.moto_api._internal import mock_random
+from moto.utilities.utils import get_partition
 from moto.workspaces.exceptions import (
     InvalidParameterValuesException,
     ResourceAlreadyExistsException,
@@ -141,7 +142,9 @@ class WorkSpaceDirectory(BaseModel):
         self.subnet_ids = subnet_ids or dir_subnet_ids
         self.dns_ip_addresses = directory.dns_ip_addrs
         self.customer_username = "Administrator"
-        self.iam_rold_id = f"arn:aws:iam::{account_id}:role/workspaces_DefaultRole"
+        self.iam_rold_id = (
+            f"arn:{get_partition(region)}:iam::{account_id}:role/workspaces_DefaultRole"
+        )
         dir_type = directory.directory_type
         if dir_type == "ADConnector":
             self.directory_type = "AD_CONNECTOR"


Test patch:
diff --git a/tests/test_acmpca/test_acmpca.py b/tests/test_acmpca/test_acmpca.py
index e68345fc9..bbcc4c071 100644
--- a/tests/test_acmpca/test_acmpca.py
+++ b/tests/test_acmpca/test_acmpca.py
@@ -18,8 +18,11 @@ from moto.core.utils import utcnow
 
 
 @mock_aws
-def test_create_certificate_authority():
-    client = boto3.client("acm-pca", region_name="eu-west-1")
+@pytest.mark.parametrize(
+    "region,partition", [("eu-west-1", "aws"), ("us-gov-east-1", "aws-us-gov")]
+)
+def test_create_certificate_authority(region, partition):
+    client = boto3.client("acm-pca", region_name=region)
     resp = client.create_certificate_authority(
         CertificateAuthorityConfiguration={
             "KeyAlgorithm": "RSA_4096",
@@ -31,7 +34,7 @@ def test_create_certificate_authority():
     )
 
     assert (
-        f"arn:aws:acm-pca:eu-west-1:{DEFAULT_ACCOUNT_ID}:certificate-authority/"
+        f"arn:{partition}:acm-pca:{region}:{DEFAULT_ACCOUNT_ID}:certificate-authority/"
         in resp["CertificateAuthorityArn"]
     )
 
@@ -276,8 +279,11 @@ def test_delete_certificate_authority():
 
 
 @mock_aws
-def test_issue_certificate():
-    client = boto3.client("acm-pca", region_name="ap-southeast-1")
+@pytest.mark.parametrize(
+    "region,partition", [("ap-southeast-1", "aws"), ("us-gov-east-1", "aws-us-gov")]
+)
+def test_issue_certificate(region, partition):
+    client = boto3.client("acm-pca", region_name=region)
     ca_arn = client.create_certificate_authority(
         CertificateAuthorityConfiguration={
             "KeyAlgorithm": "RSA_4096",
@@ -296,7 +302,9 @@ def test_issue_certificate():
         Validity={"Type": "YEARS", "Value": 10},
     )
 
-    assert "CertificateArn" in resp
+    assert resp["CertificateArn"].startswith(
+        f"arn:{partition}:acm-pca:{region}:{DEFAULT_ACCOUNT_ID}:certificate-authority"
+    )
 
 
 @mock_aws
diff --git a/tests/test_appsync/test_appsync_schema.py b/tests/test_appsync/test_appsync_schema.py
index 5ffc6e972..b7e52f6e5 100644
--- a/tests/test_appsync/test_appsync_schema.py
+++ b/tests/test_appsync/test_appsync_schema.py
@@ -95,8 +95,11 @@ def test_get_schema_creation_status_invalid():
 
 
 @mock_aws
-def test_get_type_from_schema():
-    client = boto3.client("appsync", region_name="us-east-2")
+@pytest.mark.parametrize(
+    "region,partition", [("us-east-2", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_get_type_from_schema(region, partition):
+    client = boto3.client("appsync", region_name=region)
 
     api_id = client.create_graphql_api(name="api1", authenticationType="API_KEY")[
         "graphqlApi"
@@ -109,7 +112,7 @@ def test_get_type_from_schema():
     graphql_type = resp["type"]
     assert graphql_type["name"] == "Post"
     assert graphql_type["description"] == "My custom post type"
-    assert graphql_type["arn"] == "arn:aws:appsync:graphql_type/Post"
+    assert graphql_type["arn"] == f"arn:{partition}:appsync:graphql_type/Post"
     assert graphql_type["definition"] == "NotYetImplemented"
     assert graphql_type["format"] == "SDL"
 
diff --git a/tests/test_ce/test_ce.py b/tests/test_ce/test_ce.py
index 8989b605a..c1044fbf9 100644
--- a/tests/test_ce/test_ce.py
+++ b/tests/test_ce/test_ce.py
@@ -1,3 +1,5 @@
+from unittest import mock
+
 import boto3
 import pytest
 from botocore.exceptions import ClientError
@@ -10,8 +12,13 @@ from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
 
 
 @mock_aws
-def test_create_cost_category_definition():
-    client = boto3.client("ce", region_name="ap-southeast-1")
+@mock.patch.dict("os.environ", {"MOTO_ENABLE_ISO_REGIONS": "true"})
+@pytest.mark.parametrize(
+    "region,partition",
+    [("us-east-1", "aws"), ("cn-north-1", "aws-cn"), ("us-isob-east-1", "aws-iso-b")],
+)
+def test_create_cost_category_definition(region, partition):
+    client = boto3.client("ce", region_name=region)
     resp = client.create_cost_category_definition(
         Name="ccd",
         RuleVersion="CostCategoryExpression.v1",
@@ -19,7 +26,9 @@ def test_create_cost_category_definition():
             {"Value": "v", "Rule": {"CostCategories": {"Key": "k", "Values": ["v"]}}}
         ],
     )
-    assert resp["CostCategoryArn"].startswith(f"arn:aws:ce::{ACCOUNT_ID}:costcategory/")
+    assert resp["CostCategoryArn"].startswith(
+        f"arn:{partition}:ce::{ACCOUNT_ID}:costcategory/"
+    )
     assert "EffectiveStart" in resp
 
 
diff --git a/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py b/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
index 46b22946c..e5c8ff28e 100644
--- a/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
+++ b/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
@@ -487,9 +487,17 @@ def test_stop_stack_set_operation():
 
 
 @mock_aws
-def test_describe_stack_set_operation():
-    cf = boto3.client("cloudformation", region_name=REGION_NAME)
-    cf.create_stack_set(StackSetName="name", TemplateBody=dummy_template_json)
+@pytest.mark.parametrize(
+    "region,partition", [(REGION_NAME, "aws"), ("cn-north-1", "aws-cn")]
+)
+@pytest.mark.parametrize("include_role", [True, False])
+def test_describe_stack_set_operation(region, partition, include_role):
+    cf = boto3.client("cloudformation", region_name=region)
+    kwargs = (
+        {"AdministrationRoleARN": "arn:my_role_with_long_name"} if include_role else {}
+    )
+    cf.create_stack_set(StackSetName="name", TemplateBody=dummy_template_json, **kwargs)
+
     operation_id = cf.create_stack_instances(
         StackSetName="name",
         Accounts=[ACCOUNT_ID],
@@ -497,12 +505,20 @@ def test_describe_stack_set_operation():
     )["OperationId"]
 
     cf.stop_stack_set_operation(StackSetName="name", OperationId=operation_id)
-    response = cf.describe_stack_set_operation(
+    stack_set_op = cf.describe_stack_set_operation(
         StackSetName="name", OperationId=operation_id
-    )
+    )["StackSetOperation"]
+
+    if include_role:
+        assert stack_set_op["AdministrationRoleARN"] == "arn:my_role_with_long_name"
+    else:
+        assert (
+            stack_set_op["AdministrationRoleARN"]
+            == f"arn:{partition}:iam::123456789012:role/AWSCloudFormationStackSetAdministrationRole"
+        )
+    assert stack_set_op["Status"] == "STOPPED"
+    assert stack_set_op["Action"] == "CREATE"
 
-    assert response["StackSetOperation"]["Status"] == "STOPPED"
-    assert response["StackSetOperation"]["Action"] == "CREATE"
     with pytest.raises(ClientError) as exp:
         cf.describe_stack_set_operation(
             StackSetName="name", OperationId="non_existing_operation"
@@ -851,6 +867,21 @@ def test_create_stack_set():
     assert stack_set["Description"] == "desc"
 
 
+@mock_aws
+@pytest.mark.parametrize(
+    "region,partition", [(REGION_NAME, "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_create_stack_set__without_admin_role(region, partition):
+    cf = boto3.client("cloudformation", region_name=region)
+    cf.create_stack_set(StackSetName="teststackset", TemplateBody=dummy_template_json)
+
+    stack_set = cf.describe_stack_set(StackSetName="teststackset")["StackSet"]
+    assert (
+        stack_set["AdministrationRoleARN"]
+        == f"arn:{partition}:iam::{ACCOUNT_ID}:role/AWSCloudFormationStackSetAdministrationRole"
+    )
+
+
 @mock_aws
 @pytest.mark.parametrize("name", ["1234", "stack_set", "-set"])
 def test_create_stack_set__invalid_name(name):
diff --git a/tests/test_cloudfront/test_cloudfront_distributions.py b/tests/test_cloudfront/test_cloudfront_distributions.py
index 2ac78bc20..1fa488207 100644
--- a/tests/test_cloudfront/test_cloudfront_distributions.py
+++ b/tests/test_cloudfront/test_cloudfront_distributions.py
@@ -9,12 +9,18 @@ from . import cloudfront_test_scaffolding as scaffold
 
 
 @mock_aws
-def test_create_distribution_s3_minimum():
-    client = boto3.client("cloudfront", region_name="us-west-1")
+@pytest.mark.parametrize(
+    "region,partition", [("us-west-1", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_create_distribution_s3_minimum(region, partition):
+    client = boto3.client("cloudfront", region_name=region)
     config = scaffold.example_distribution_config("ref")
 
     dist = client.create_distribution(DistributionConfig=config)["Distribution"]
-    assert dist["ARN"] == f"arn:aws:cloudfront:{ACCOUNT_ID}:distribution/{dist['Id']}"
+    assert (
+        dist["ARN"]
+        == f"arn:{partition}:cloudfront:{ACCOUNT_ID}:distribution/{dist['Id']}"
+    )
     assert dist["Status"] == "InProgress"
     assert "LastModifiedTime" in dist
     assert dist["InProgressInvalidationBatches"] == 0
diff --git a/tests/test_cloudwatch/test_cloudwatch_alarms.py b/tests/test_cloudwatch/test_cloudwatch_alarms.py
index f8fd9d016..aa5f93842 100644
--- a/tests/test_cloudwatch/test_cloudwatch_alarms.py
+++ b/tests/test_cloudwatch/test_cloudwatch_alarms.py
@@ -1,12 +1,15 @@
 import boto3
+import pytest
 
 from moto import mock_aws
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
 
 
 @mock_aws
-def test_create_alarm():
-    region = "eu-west-1"
+@pytest.mark.parametrize(
+    "region,partition", [("eu-west-1", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_create_alarm(region, partition):
     cloudwatch = boto3.client("cloudwatch", region)
 
     name = "tester"
@@ -44,7 +47,10 @@ def test_create_alarm():
     assert alarm["OKActions"] == ["arn:ok"]
     assert alarm["InsufficientDataActions"] == ["arn:insufficient"]
     assert alarm["Unit"] == "Seconds"
-    assert alarm["AlarmArn"] == f"arn:aws:cloudwatch:{region}:{ACCOUNT_ID}:alarm:{name}"
+    assert (
+        alarm["AlarmArn"]
+        == f"arn:{partition}:cloudwatch:{region}:{ACCOUNT_ID}:alarm:{name}"
+    )
     # default value should be True
     assert alarm["ActionsEnabled"] is True
 
diff --git a/tests/test_cloudwatch/test_cloudwatch_dashboards.py b/tests/test_cloudwatch/test_cloudwatch_dashboards.py
index 80d3f79ca..2933a38ab 100644
--- a/tests/test_cloudwatch/test_cloudwatch_dashboards.py
+++ b/tests/test_cloudwatch/test_cloudwatch_dashboards.py
@@ -6,14 +6,21 @@ from moto import mock_aws
 
 
 @mock_aws
-def test_put_list_dashboard():
-    client = boto3.client("cloudwatch", region_name="eu-central-1")
+@pytest.mark.parametrize(
+    "region,partition", [("eu-central-1", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_put_list_dashboard(region, partition):
+    client = boto3.client("cloudwatch", region_name=region)
     widget = '{"widgets": [{"type": "text", "x": 0, "y": 7, "width": 3, "height": 3, "properties": {"markdown": "Hello world"}}]}'
 
     client.put_dashboard(DashboardName="test1", DashboardBody=widget)
-    resp = client.list_dashboards()
+    dashboards = client.list_dashboards()["DashboardEntries"]
 
-    assert len(resp["DashboardEntries"]) == 1
+    assert len(dashboards) == 1
+    assert (
+        dashboards[0]["DashboardArn"]
+        == f"arn:{partition}:cloudwatch::123456789012:dashboard/test1"
+    )
 
 
 @mock_aws
diff --git a/tests/test_core/test_auth.py b/tests/test_core/test_auth.py
index 9dedf9040..d4b61888c 100644
--- a/tests/test_core/test_auth.py
+++ b/tests/test_core/test_auth.py
@@ -25,9 +25,12 @@ def create_user_with_access_key(user_name: str = "test-user") -> Dict[str, str]:
 
 @mock_aws
 def create_user_with_access_key_and_inline_policy(  # type: ignore[misc]
-    user_name: str, policy_document: Dict[str, Any], policy_name: str = "policy1"
+    user_name: str,
+    policy_document: Dict[str, Any],
+    policy_name: str = "policy1",
+    region_name: str = "us-east-1",
 ) -> Dict[str, str]:
-    client = boto3.client("iam", region_name="us-east-1")
+    client = boto3.client("iam", region_name=region_name)
     client.create_user(UserName=user_name)
     client.put_user_policy(
         UserName=user_name,
@@ -822,7 +825,10 @@ def test_s3_invalid_token_with_temporary_credentials() -> None:
 
 @set_initial_no_auth_action_count(3)
 @mock_aws
-def test_allow_bucket_access_using_resource_arn() -> None:
+@pytest.mark.parametrize(
+    "region,partition", [("us-west-2", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_allow_bucket_access_using_resource_arn(region: str, partition: str) -> None:
     user_name = "test-user"
     policy_doc = {
         "Version": "2012-10-17",
@@ -830,21 +836,25 @@ def test_allow_bucket_access_using_resource_arn() -> None:
             {
                 "Action": ["s3:*"],
                 "Effect": "Allow",
-                "Resource": "arn:aws:s3:::my_bucket",
+                "Resource": f"arn:{partition}:s3:::my_bucket",
                 "Sid": "BucketLevelGrants",
             },
         ],
     }
-    access_key = create_user_with_access_key_and_inline_policy(user_name, policy_doc)
+    access_key = create_user_with_access_key_and_inline_policy(
+        user_name, policy_doc, region_name=region
+    )
 
     s3_client = boto3.client(
         "s3",
-        region_name="us-east-1",
+        region_name=region,
         aws_access_key_id=access_key["AccessKeyId"],
         aws_secret_access_key=access_key["SecretAccessKey"],
     )
 
-    s3_client.create_bucket(Bucket="my_bucket")
+    s3_client.create_bucket(
+        Bucket="my_bucket", CreateBucketConfiguration={"LocationConstraint": region}
+    )
     with pytest.raises(ClientError):
         s3_client.create_bucket(Bucket="my_bucket2")
 
diff --git a/tests/test_iam/test_iam.py b/tests/test_iam/test_iam.py
index 5da763138..9b00a7dbe 100644
--- a/tests/test_iam/test_iam.py
+++ b/tests/test_iam/test_iam.py
@@ -706,11 +706,14 @@ def test_get_policy():
 
 
 @mock_aws(config={"iam": {"load_aws_managed_policies": True}})
-def test_get_aws_managed_policy():
+@pytest.mark.parametrize(
+    "region,partition", [("us-west-2", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_get_aws_managed_policy(region, partition):
     if settings.TEST_SERVER_MODE:
         raise SkipTest("Policies not loaded in ServerMode")
-    conn = boto3.client("iam", region_name="us-east-1")
-    managed_policy_arn = "arn:aws:iam::aws:policy/IAMUserChangePassword"
+    conn = boto3.client("iam", region_name=region)
+    managed_policy_arn = f"arn:{partition}:iam::aws:policy/IAMUserChangePassword"
     managed_policy_create_date = datetime.strptime(
         "2016-11-15T00:25:16+00:00", "%Y-%m-%dT%H:%M:%S+00:00"
     )
@@ -3158,9 +3161,12 @@ def test_create_role_no_path():
 
 
 @mock_aws()
-def test_create_role_with_permissions_boundary():
-    conn = boto3.client("iam", region_name="us-east-1")
-    boundary = f"arn:aws:iam::{ACCOUNT_ID}:policy/boundary"
+@pytest.mark.parametrize(
+    "region,partition", [("us-west-2", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_create_role_with_permissions_boundary(region, partition):
+    conn = boto3.client("iam", region_name=region)
+    boundary = f"arn:{partition}:iam::{ACCOUNT_ID}:policy/boundary"
     resp = conn.create_role(
         RoleName="my-role",
         AssumeRolePolicyDocument="some policy",
@@ -3182,10 +3188,15 @@ def test_create_role_with_permissions_boundary():
 
     invalid_boundary_arn = "arn:aws:iam::123456789:not_a_boundary"
 
-    with pytest.raises(ClientError):
+    with pytest.raises(ClientError) as exc:
         conn.put_role_permissions_boundary(
             RoleName="my-role", PermissionsBoundary=invalid_boundary_arn
         )
+    err = exc.value.response["Error"]
+    assert (
+        err["Message"]
+        == "Value (arn:aws:iam::123456789:not_a_boundary) for parameter PermissionsBoundary is invalid."
+    )
 
     with pytest.raises(ClientError):
         conn.create_role(
diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py
index ed3a52cd6..8c1184e3a 100644
--- a/tests/test_s3/test_s3.py
+++ b/tests/test_s3/test_s3.py
@@ -35,8 +35,8 @@ class MyModel:
         self.value = value
         self.metadata = metadata or {}
 
-    def save(self):
-        s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    def save(self, region=None):
+        s3_client = boto3.client("s3", region_name=region or DEFAULT_REGION_NAME)
         s3_client.put_object(
             Bucket="mybucket", Key=self.name, Body=self.value, Metadata=self.metadata
         )
@@ -57,14 +57,20 @@ def test_keys_are_pickleable():
 
 
 @mock_aws
-def test_my_model_save():
+@pytest.mark.parametrize(
+    "region,partition",
+    [("us-west-2", "aws"), ("cn-north-1", "aws-cn"), ("us-isob-east-1", "aws-iso-b")],
+)
+def test_my_model_save(region, partition):
     # Create Bucket so that test can run
-    conn = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
-    conn.create_bucket(Bucket="mybucket")
+    conn = boto3.resource("s3", region_name=region)
+    conn.create_bucket(
+        Bucket="mybucket", CreateBucketConfiguration={"LocationConstraint": region}
+    )
     ####################################
 
     model_instance = MyModel("steve", "is awesome")
-    model_instance.save()
+    model_instance.save(region)
 
     body = conn.Object("mybucket", "steve").get()["Body"].read().decode()
 
@@ -522,11 +528,17 @@ def test_bucket_key_listing_order():
 
 
 @mock_aws
-def test_key_with_reduced_redundancy():
-    s3_resource = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
+@pytest.mark.parametrize(
+    "region,partition",
+    [("us-west-2", "aws"), ("cn-north-1", "aws-cn"), ("us-isob-east-1", "aws-iso-b")],
+)
+def test_key_with_reduced_redundancy(region, partition):
+    s3_resource = boto3.resource("s3", region_name=region)
     bucket_name = "test_bucket"
+    s3_resource.create_bucket(
+        Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": region}
+    )
     bucket = s3_resource.Bucket(bucket_name)
-    bucket.create()
 
     bucket.put_object(
         Key="test_rr_key", Body=b"somedata", StorageClass="REDUCED_REDUNDANCY"
@@ -2194,9 +2206,14 @@ def test_delete_bucket_cors(bucket_name=None):
 
 
 @mock_aws
-def test_put_bucket_notification():
-    s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
-    s3_client.create_bucket(Bucket="bucket")
+@pytest.mark.parametrize(
+    "region,partition", [("us-west-2", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_put_bucket_notification(region, partition):
+    s3_client = boto3.client("s3", region_name=region)
+    s3_client.create_bucket(
+        Bucket="bucket", CreateBucketConfiguration={"LocationConstraint": region}
+    )
 
     # With no configuration:
     result = s3_client.get_bucket_notification(Bucket="bucket")
@@ -2210,11 +2227,11 @@ def test_put_bucket_notification():
         NotificationConfiguration={
             "TopicConfigurations": [
                 {
-                    "TopicArn": "arn:aws:sns:us-east-1:012345678910:mytopic",
+                    "TopicArn": f"arn:{partition}:sns:{region}:012345678910:mytopic",
                     "Events": ["s3:ObjectCreated:*", "s3:ObjectRemoved:*"],
                 },
                 {
-                    "TopicArn": "arn:aws:sns:us-east-1:012345678910:myothertopic",
+                    "TopicArn": f"arn:{partition}:sns:{region}:012345678910:myothertopic",
                     "Events": ["s3:ObjectCreated:*"],
                     "Filter": {
                         "Key": {
@@ -2236,11 +2253,11 @@ def test_put_bucket_notification():
     assert not result.get("LambdaFunctionConfigurations")
     assert (
         result["TopicConfigurations"][0]["TopicArn"]
-        == "arn:aws:sns:us-east-1:012345678910:mytopic"
+        == f"arn:{partition}:sns:{region}:012345678910:mytopic"
     )
     assert (
         result["TopicConfigurations"][1]["TopicArn"]
-        == "arn:aws:sns:us-east-1:012345678910:myothertopic"
+        == f"arn:{partition}:sns:{region}:012345678910:myothertopic"
     )
     assert len(result["TopicConfigurations"][0]["Events"]) == 2
     assert len(result["TopicConfigurations"][1]["Events"]) == 1
@@ -2275,7 +2292,7 @@ def test_put_bucket_notification():
             "QueueConfigurations": [
                 {
                     "Id": "SomeID",
-                    "QueueArn": "arn:aws:sqs:us-east-1:012345678910:myQueue",
+                    "QueueArn": f"arn:{partition}:sqs:{region}:012345678910:myQueue",
                     "Events": ["s3:ObjectCreated:*"],
                     "Filter": {
                         "Key": {"FilterRules": [{"Name": "prefix", "Value": "images/"}]}
@@ -2291,7 +2308,7 @@ def test_put_bucket_notification():
     assert result["QueueConfigurations"][0]["Id"] == "SomeID"
     assert (
         result["QueueConfigurations"][0]["QueueArn"]
-        == "arn:aws:sqs:us-east-1:012345678910:myQueue"
+        == f"arn:{partition}:sqs:{region}:012345678910:myQueue"
     )
     assert result["QueueConfigurations"][0]["Events"][0] == "s3:ObjectCreated:*"
     assert len(result["QueueConfigurations"][0]["Events"]) == 1
@@ -2311,7 +2328,7 @@ def test_put_bucket_notification():
         NotificationConfiguration={
             "LambdaFunctionConfigurations": [
                 {
-                    "LambdaFunctionArn": "arn:aws:lambda:us-east-1:012345678910:function:lambda",
+                    "LambdaFunctionArn": f"arn:{partition}:lambda:{region}:012345678910:function:lambda",
                     "Events": ["s3:ObjectCreated:*"],
                     "Filter": {
                         "Key": {"FilterRules": [{"Name": "prefix", "Value": "images/"}]}
@@ -2327,7 +2344,7 @@ def test_put_bucket_notification():
     assert result["LambdaFunctionConfigurations"][0]["Id"]
     assert (
         result["LambdaFunctionConfigurations"][0]["LambdaFunctionArn"]
-        == "arn:aws:lambda:us-east-1:012345678910:function:lambda"
+        == f"arn:{partition}:lambda:{region}:012345678910:function:lambda"
     )
     assert (
         result["LambdaFunctionConfigurations"][0]["Events"][0] == "s3:ObjectCreated:*"
@@ -2356,19 +2373,19 @@ def test_put_bucket_notification():
         NotificationConfiguration={
             "TopicConfigurations": [
                 {
-                    "TopicArn": "arn:aws:sns:us-east-1:012345678910:mytopic",
+                    "TopicArn": f"arn:{partition}:sns:{region}:012345678910:mytopic",
                     "Events": ["s3:ObjectCreated:*", "s3:ObjectRemoved:*"],
                 }
             ],
             "LambdaFunctionConfigurations": [
                 {
-                    "LambdaFunctionArn": "arn:aws:lambda:us-east-1:012345678910:function:lambda",
+                    "LambdaFunctionArn": f"arn:{partition}:lambda:{region}:012345678910:function:lambda",
                     "Events": ["s3:ObjectCreated:*"],
                 }
             ],
             "QueueConfigurations": [
                 {
-                    "QueueArn": "arn:aws:sqs:us-east-1:012345678910:myQueue",
+                    "QueueArn": f"arn:{partition}:sqs:{region}:012345678910:myQueue",
                     "Events": ["s3:ObjectCreated:*"],
                 }
             ],
diff --git a/tests/test_s3/test_s3_eventbridge_integration.py b/tests/test_s3/test_s3_eventbridge_integration.py
index 8c1e3772f..df7c2255a 100644
--- a/tests/test_s3/test_s3_eventbridge_integration.py
+++ b/tests/test_s3/test_s3_eventbridge_integration.py
@@ -5,6 +5,7 @@ from unittest import SkipTest
 from uuid import uuid4
 
 import boto3
+import pytest
 import requests
 
 from moto import mock_aws, settings
@@ -24,34 +25,40 @@ def _seteup_bucket_notification_eventbridge(
     bucket_name: str = str(uuid4()),
     rule_name: str = "test-rule",
     log_group_name: str = "/test-group",
+    region: str = REGION_NAME,
 ) -> Dict[str, str]:
     """Setups S3, EventBridge and CloudWatchLogs"""
     # Setup S3
-    s3_res = boto3.resource("s3", region_name=REGION_NAME)
-    s3_res.create_bucket(Bucket=bucket_name)
+    s3_res = boto3.resource("s3", region_name=region)
+    if region == "us-east-1":
+        s3_res.create_bucket(Bucket=bucket_name)
+    else:
+        s3_res.create_bucket(
+            Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": region}
+        )
 
     # Put bucket notification event bridge
-    s3_client = boto3.client("s3", region_name=REGION_NAME)
+    s3_client = boto3.client("s3", region_name=region)
     s3_client.put_bucket_notification_configuration(
         Bucket=bucket_name,
         NotificationConfiguration={"EventBridgeConfiguration": {}},
     )
 
     # Setup EventBridge Rule
-    events_client = boto3.client("events", region_name=REGION_NAME)
+    events_client = boto3.client("events", region_name=region)
     events_client.put_rule(
         Name=rule_name, EventPattern=json.dumps({"account": [ACCOUNT_ID]})
     )
 
     # Create a log group and attach it to the events target.
-    logs_client = boto3.client("logs", region_name=REGION_NAME)
+    logs_client = boto3.client("logs", region_name=region)
     logs_client.create_log_group(logGroupName=log_group_name)
     events_client.put_targets(
         Rule=rule_name,
         Targets=[
             {
                 "Id": "test",
-                "Arn": f"arn:aws:logs:{REGION_NAME}:{ACCOUNT_ID}:log-group:{log_group_name}",
+                "Arn": f"arn:aws:logs:{region}:{ACCOUNT_ID}:log-group:{log_group_name}",
             }
         ],
     )
@@ -63,8 +70,10 @@ def _seteup_bucket_notification_eventbridge(
     }
 
 
-def _get_send_events(log_group_name: str = "/test-group") -> List[Dict[str, Any]]:
-    logs_client = boto3.client("logs", region_name=REGION_NAME)
+def _get_send_events(
+    log_group_name: str = "/test-group", region: str = REGION_NAME
+) -> List[Dict[str, Any]]:
+    logs_client = boto3.client("logs", region_name=region)
     return sorted(
         logs_client.filter_log_events(logGroupName=log_group_name)["events"],
         key=lambda item: item["timestamp"],
@@ -72,21 +81,25 @@ def _get_send_events(log_group_name: str = "/test-group") -> List[Dict[str, Any]
 
 
 @mock_aws
-def test_put_object_notification_ObjectCreated_PUT():
-    resource_names = _seteup_bucket_notification_eventbridge()
+@pytest.mark.parametrize(
+    "region,partition", [("us-west-2", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_put_object_notification_ObjectCreated_PUT(region, partition):
+    resource_names = _seteup_bucket_notification_eventbridge(region=region)
     bucket_name = resource_names["bucket_name"]
-    s3_client = boto3.client("s3", region_name=REGION_NAME)
+    s3_client = boto3.client("s3", region_name=region)
 
     # Put Object
     s3_client.put_object(Bucket=bucket_name, Key="keyname", Body="bodyofnewobject")
 
-    events = _get_send_events()
+    events = _get_send_events(region=region)
     assert len(events) == 2
     event_message = json.loads(events[0]["message"])
     assert event_message["detail-type"] == "Object Created"
     assert event_message["source"] == "aws.s3"
     assert event_message["account"] == ACCOUNT_ID
-    assert event_message["region"] == REGION_NAME
+    assert event_message["region"] == region
+    assert event_message["resources"] == [f"arn:{partition}:s3:::{bucket_name}"]
     assert event_message["detail"]["bucket"]["name"] == bucket_name
     assert event_message["detail"]["reason"] == "ObjectCreated"
 
diff --git a/tests/test_s3/test_s3_lambda_integration.py b/tests/test_s3/test_s3_lambda_integration.py
index 08859f550..7f2f79061 100644
--- a/tests/test_s3/test_s3_lambda_integration.py
+++ b/tests/test_s3/test_s3_lambda_integration.py
@@ -282,17 +282,22 @@ def test_object_put__sends_to_queue__using_filter():
 
 
 @mock_aws
-def test_put_bucket_notification_sns_sqs():
-    s3_client = boto3.client("s3", region_name=REGION_NAME)
-    s3_client.create_bucket(Bucket="bucket")
+@pytest.mark.parametrize(
+    "region,partition", [("us-west-2", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_put_bucket_notification_sns_sqs(region, partition):
+    s3_client = boto3.client("s3", region_name=region)
+    s3_client.create_bucket(
+        Bucket="bucket", CreateBucketConfiguration={"LocationConstraint": region}
+    )
 
-    sqs_client = boto3.client("sqs", region_name=REGION_NAME)
+    sqs_client = boto3.client("sqs", region_name=region)
     sqs_queue = sqs_client.create_queue(QueueName="queue")
     sqs_queue_arn = sqs_client.get_queue_attributes(
         QueueUrl=sqs_queue["QueueUrl"], AttributeNames=["QueueArn"]
     )
 
-    sns_client = boto3.client("sns", region_name=REGION_NAME)
+    sns_client = boto3.client("sns", region_name=region)
     sns_topic = sns_client.create_topic(Name="topic")
 
     # Subscribe SQS queue to SNS topic
@@ -352,6 +357,11 @@ def test_put_bucket_notification_sns_sqs():
     # Get S3 notification from SNS message
     s3_message_body = json.loads(sns_message["Message"])
     assert s3_message_body["Records"][0]["eventName"] == "ObjectCreated:Put"
+    assert s3_message_body["Records"][0]["awsRegion"] == region
+    assert (
+        s3_message_body["Records"][0]["s3"]["bucket"]["arn"]
+        == f"arn:{partition}:s3:::bucket"
+    )
 
 
 @mock_aws
diff --git a/tests/test_ssoadmin/test_ssoadmin_policies.py b/tests/test_ssoadmin/test_ssoadmin_policies.py
index e3fcfd9ee..3a4085257 100644
--- a/tests/test_ssoadmin/test_ssoadmin_policies.py
+++ b/tests/test_ssoadmin/test_ssoadmin_policies.py
@@ -139,12 +139,15 @@ def test_delete_inline_policy_to_permissionset():
 
 
 @mock_aws
-def test_attach_managed_policy_to_permission_set():
-    client = boto3.client("sso-admin", region_name="us-east-1")
+@pytest.mark.parametrize(
+    "region,partition", [("us-east-1", "aws"), ("cn-north-1", "aws-cn")]
+)
+def test_attach_managed_policy_to_permission_set(region, partition):
+    client = boto3.client("sso-admin", region_name=region)
 
     permission_set_arn = create_permissionset(client)
     permissionset_id = permission_set_arn.split("/")[-1]
-    managed_policy_arn = "arn:aws:iam::aws:policy/AdministratorAccess"
+    managed_policy_arn = f"arn:{partition}:iam::aws:policy/AdministratorAccess"
 
     client.attach_managed_policy_to_permission_set(
         InstanceArn=DUMMY_INSTANCE_ARN,
@@ -158,10 +161,7 @@ def test_attach_managed_policy_to_permission_set():
     )
 
     assert response["AttachedManagedPolicies"][0]["Name"] == "AdministratorAccess"
-    assert (
-        response["AttachedManagedPolicies"][0]["Arn"]
-        == "arn:aws:iam::aws:policy/AdministratorAccess"
-    )
+    assert response["AttachedManagedPolicies"][0]["Arn"] == managed_policy_arn
 
     # test for managed policy that is already attached
     with pytest.raises(ClientError) as e:
diff --git a/tests/test_utilities/test_utils.py b/tests/test_utilities/test_utils.py
new file mode 100644
index 000000000..3b76529a4
--- /dev/null
+++ b/tests/test_utilities/test_utils.py
@@ -0,0 +1,16 @@
+import re
+
+from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition
+
+
+def test_get_partition():
+    assert get_partition(None) == "aws"
+    assert get_partition("unknown") == "aws"
+    assert get_partition("cn-north-1") == "aws-cn"
+
+
+def test_partition_regex():
+    assert re.match(ARN_PARTITION_REGEX, "arn:aws:someservice")
+    assert re.match(ARN_PARTITION_REGEX, "arn:aws-cn:someservice")
+    assert not re.match(ARN_PARTITION_REGEX, "arn:partition:someservice")
+    assert not re.match(ARN_PARTITION_REGEX, "someservice")
diff --git a/tests/test_wafv2/test_wafv2.py b/tests/test_wafv2/test_wafv2.py
index 54ba198be..0646d2fe1 100644
--- a/tests/test_wafv2/test_wafv2.py
+++ b/tests/test_wafv2/test_wafv2.py
@@ -9,13 +9,17 @@ from .test_helper_functions import CREATE_WEB_ACL_BODY, LIST_WEB_ACL_BODY
 
 
 @mock_aws
-def test_create_web_acl():
-    conn = boto3.client("wafv2", region_name="us-east-1")
+@pytest.mark.parametrize(
+    "region,partition",
+    [("us-east-1", "aws"), ("cn-north-1", "aws-cn"), ("us-gov-east-1", "aws-us-gov")],
+)
+def test_create_web_acl(region, partition):
+    conn = boto3.client("wafv2", region_name=region)
     res = conn.create_web_acl(**CREATE_WEB_ACL_BODY("John", "REGIONAL"))
     web_acl = res["Summary"]
     assert web_acl.get("Name") == "John"
     assert web_acl.get("ARN").startswith(
-        f"arn:aws:wafv2:us-east-1:{ACCOUNT_ID}:regional/webacl/John/"
+        f"arn:{partition}:wafv2:{region}:{ACCOUNT_ID}:regional/webacl/John/"
     )
     # Duplicate name - should raise error
     with pytest.raises(ClientError) as ex: