Partitions: Regex validation and missing ARN's (#7716)

moto/acmpca/models.py

@@ -15,6 +15,7 @@
 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 @@ def __init__(
         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 @@ def issue_certificate(self, csr_bytes: bytes) -> str:
         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
 

moto/apigateway/models.py

@@ -23,6 +23,7 @@
 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 @@ def put_integration(
     ) -> 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 @@ def put_integration(
         ]:
             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(

moto/apigateway/urls.py

@@ -11,7 +11,7 @@
     "{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,

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.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(":")

moto/appsync/models.py

@@ -41,8 +41,9 @@
 
 
 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 @@ def get_type(self, name: str) -> Optional[Dict[str, Any]]:  # type: ignore[retur
                     "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 @@ def update_api_key(
     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]

moto/autoscaling/models.py

@@ -449,7 +449,8 @@ def __init__(
         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)

moto/awslambda/models.py

@@ -40,7 +40,11 @@
 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 @@ def get_function_by_name_or_arn_forbid_qualifier(
         :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 @@ def get_function_by_name_or_arn_with_qualifier(
         :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 @@ def get_function_by_name_or_arn_with_qualifier(
     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 @@ def publish_version(
 
     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]

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 @@ def _set_configuration_qualifier(  # type: ignore[misc]
         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]

moto/awslambda/utils.py

@@ -2,7 +2,7 @@
 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 @@ def make_ver_arn(
 
 
 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(":")
 

moto/backend_index.py

@@ -21,12 +21,13 @@
     ("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")),

moto/ce/models.py

@@ -8,7 +8,7 @@
 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 @@ def __init__(
         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 @@ def create_cost_category_definition(
         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)

moto/cloudformation/models.py

@@ -55,8 +55,10 @@ def __init__(
         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(

moto/cloudformation/responses.py

@@ -531,8 +531,6 @@ def describe_stack_set(self) -> str:
         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 @@ def set_stack_policy(self) -> str:
   <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>

moto/cloudfront/models.py

@@ -7,7 +7,7 @@
 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 @@ def random_id(uppercase: bool = True) -> str:
         )
         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 @@ def create_distribution(
     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:

moto/cloudfront/urls.py

@@ -4,6 +4,7 @@
 
 url_bases = [
     r"https?://cloudfront\.amazonaws\.com",
+    r"https?://cloudfront\.(.+)\.amazonaws\.com",
 ]
 url_paths = {
     "{0}/2020-05-31/distribution$": CloudFrontResponse.dispatch,

moto/cloudwatch/models.py

@@ -304,9 +304,9 @@ def __init__(
 
 
 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 @@ def get_all_metrics(self) -> List[MetricDatumBase]:
         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():