localstack · baermat · Feb 5, 2026 · Feb 4, 2026 · Feb 4, 2026 · Feb 4, 2026
@@ -211,7 +211,6 @@ jobs:
       dynamodb-v2: ${{ steps.changes.outputs.dynamodb-v2 }}
       events-v1: ${{ steps.changes.outputs.events-v1 }}
       cloudformation-legacy: ${{ steps.changes.outputs.cloudformation-legacy }}
-      sns-v2: ${{ steps.changes.outputs.sns-v2 }}
     steps:
       - name: Checkout
         uses: actions/checkout@v6
@@ -280,8 +279,6 @@ jobs:
             cloudformation-legacy:
                - 'localstack-core/localstack/services/cloudformation/**'
                - 'tests/aws/services/cloudformation/**'
-            sns-v2:
-               - 'tests/aws/services/sns/**' # todo: potentially add more locations (lambda/sqs tests?)
 
       - name: Run Unit Tests
         timeout-minutes: 20
@@ -868,59 +865,6 @@ jobs:
             ${{ env.JUNIT_REPORTS_FILE }}
           retention-days: 30
 
-  test-sns-v2:
-    name: Test SNS V2
-    if: ${{ !inputs.onlyAcceptanceTests && (github.ref == 'refs/heads/main' || startsWith(github.ref, 'refs/tags/v') || needs.test-preflight.outputs.sns-v2 == 'true') }}
-    runs-on: ubuntu-latest
-    needs:
-      - test-preflight
-      - build
-    timeout-minutes: 60
-    env:
-      # Set job-specific environment variables for pytest-tinybird
-      CI_JOB_NAME: ${{ github.job }}
-      CI_JOB_ID: ${{ github.job }}
-    outputs:
-      # we need this output to conditionally execute the Publishing step
-      job_status: "executed"
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v6
-
-      - name: Prepare Local Test Environment
-        uses: ./.github/actions/setup-tests-env
-
-      - name: Download Test Selection
-        if: ${{ env.TESTSELECTION_PYTEST_ARGS }}
-        uses: actions/download-artifact@v7
-        with:
-          name: test-selection
-          path: dist/testselection/
-
-      - name: Run SNS v2 Provider Tests
-        timeout-minutes: 30
-        env:
-          # add the GitHub API token to avoid rate limit issues
-          GITHUB_API_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          DEBUG: 1
-          COVERAGE_FILE: ".coverage.sns_v2"
-          TEST_PATH: "tests/aws/services/sns/"
-          JUNIT_REPORTS_FILE: "pytest-junit-sns-v2.xml"
-          PYTEST_ARGS: "${{ env.TINYBIRD_PYTEST_ARGS }}${{ env.TESTSELECTION_PYTEST_ARGS }} --reruns 3 -o junit_suite_name=sns_v2"
-          PROVIDER_OVERRIDE_SNS: "v2"
-        run: make test-coverage
-
-      - name: Archive Test Results
-        uses: actions/upload-artifact@v6
-        if: success() || failure()
-        with:
-          name: test-results-sns-v2
-          include-hidden-files: true
-          path: |
-            pytest-junit-sns-v2.xml
-            .coverage.sns_v2
-          retention-days: 30
-
   publish-alternative-provider-test-results:
     name: Publish Alternative Provider Test Results
     # execute on success or failure, but not if the workflow is cancelled or all of the dependencies has been skipped
@@ -930,7 +874,6 @@ jobs:
       - test-events-v1
       - test-ddb-v2
       - test-cloudwatch-v1
-      - test-sns-v2
     runs-on: ubuntu-latest
     permissions:
       checks: write
@@ -958,11 +901,6 @@ jobs:
         with:
           pattern: test-results-cloudwatch-v1
 
-      - name: Download SNS v2 Artifacts
-        uses: actions/download-artifact@v7
-        with:
-          pattern: test-results-sns-v2
-
       - name: Publish Bootstrap and Integration Test Results
         uses: EnricoMi/publish-unit-test-result-action@v2
         if: success() || failure()

diff --git a/localstack-core/localstack/services/providers.py b/localstack-core/localstack/services/providers.py
@@ -311,17 +311,8 @@ def ses():
 
 @aws_provider()
 def sns():
-    from localstack.services.moto import MotoFallbackDispatcher
     from localstack.services.sns.provider import SnsProvider
 
-    provider = SnsProvider()
-    return Service.for_provider(provider, dispatch_table_factory=MotoFallbackDispatcher)
-
-
-@aws_provider(api="sns", name="v2")
-def sns_v2():
-    from localstack.services.sns.v2.provider import SnsProvider
-
     provider = SnsProvider()
     return Service.for_provider(provider)
 

diff --git a/localstack-core/localstack/services/sns/constants.py b/localstack-core/localstack/services/sns/constants.py
@@ -2,7 +2,7 @@
 from string import ascii_letters, digits
 from typing import get_args
 
-from localstack.services.sns.v2.models import SnsApplicationPlatforms
+from localstack.services.sns.models import SnsApplicationPlatforms
 
 SNS_PROTOCOLS = [
     "http",

diff --git a/localstack-core/localstack/services/sns/models.py b/localstack-core/localstack/services/sns/models.py
@@ -5,15 +5,33 @@
 from typing import Literal, TypedDict
 
 from localstack.aws.api.sns import (
+    Endpoint,
     MessageAttributeMap,
+    PhoneNumber,
+    PlatformApplication,
     PublishBatchRequestEntry,
+    TopicAttributesMap,
     subscriptionARN,
     topicARN,
 )
-from localstack.services.stores import AccountRegionBundle, BaseStore, LocalAttribute
-from localstack.utils.aws.arns import parse_arn
+from localstack.services.stores import (
+    AccountRegionBundle,
+    BaseStore,
+    CrossRegionAttribute,
+    LocalAttribute,
+)
 from localstack.utils.objects import singleton_factory
 from localstack.utils.strings import long_uid
+from localstack.utils.tagging import TaggingService
+
+
+class Topic(TypedDict, total=True):
+    arn: str
+    name: str
+    attributes: TopicAttributesMap
+    data_protection_policy: str
+    subscriptions: list[str]
+
 
 SnsProtocols = Literal[
     "http", "https", "email", "email-json", "sms", "sqs", "application", "lambda", "firehose"
@@ -23,39 +41,47 @@
     "APNS", "APNS_SANDBOX", "ADM", "FCM", "Baidu", "GCM", "MPNS", "WNS"
 ]
 
+
+class EndpointAttributeNames(StrEnum):
+    CUSTOM_USER_DATA = "CustomUserData"
+    Token = "Token"
+    ENABLED = "Enabled"
+
+
+SMS_ATTRIBUTE_NAMES = [
+    "DeliveryStatusIAMRole",
+    "DeliveryStatusSuccessSamplingRate",
+    "DefaultSenderID",
+    "DefaultSMSType",
+    "UsageReportS3Bucket",
+]
+SMS_TYPES = ["Promotional", "Transactional"]
+SMS_DEFAULT_SENDER_REGEX = r"^(?=[A-Za-z0-9]{1,11}$)(?=.*[A-Za-z])[A-Za-z0-9]+$"
 SnsMessageProtocols = Literal[SnsProtocols, SnsApplicationPlatforms]
 
 
-def create_default_sns_topic_policy(topic_arn: str) -> dict:
+class SnsSubscription(TypedDict, total=False):
     """
-    Creates the default SNS topic policy for the given topic ARN.
-
-    :param topic_arn: The topic arn
-    :return: A policy document
+    In SNS, Subscription can be represented with only TopicArn, Endpoint, Protocol, SubscriptionArn and Owner, for
+    example in ListSubscriptions. However, when getting a subscription with GetSubscriptionAttributes, it will return
+    the Subscription object merged with its own attributes.
+    This represents this merged object, for internal use and in GetSubscriptionAttributes
+    https://docs.aws.amazon.com/cli/latest/reference/sns/get-subscription-attributes.html
     """
-    return {
-        "Version": "2008-10-17",
-        "Id": "__default_policy_ID",
-        "Statement": [
-            {
-                "Sid": "__default_statement_ID",
-                "Effect": "Allow",
-                "Principal": {"AWS": "*"},
-                "Action": [
-                    "SNS:GetTopicAttributes",
-                    "SNS:SetTopicAttributes",
-                    "SNS:AddPermission",
-                    "SNS:RemovePermission",
-                    "SNS:DeleteTopic",
-                    "SNS:Subscribe",
-                    "SNS:ListSubscriptionsByTopic",
-                    "SNS:Publish",
-                ],
-                "Resource": topic_arn,
-                "Condition": {"StringEquals": {"AWS:SourceOwner": parse_arn(topic_arn)["account"]}},
-            }
-        ],
-    }
+
+    TopicArn: topicARN
+    Endpoint: str
+    Protocol: SnsProtocols
+    SubscriptionArn: subscriptionARN
+    PendingConfirmation: Literal["true", "false"]
+    Owner: str | None
+    SubscriptionPrincipal: str | None
+    FilterPolicy: str | None
+    FilterPolicyScope: Literal["MessageAttributes", "MessageBody"]
+    RawMessageDelivery: Literal["true", "false"]
+    ConfirmationWasAuthenticated: Literal["true", "false"]
+    SubscriptionRoleArn: str | None
+    DeliveryPolicy: str | None
 
 
 @singleton_factory
@@ -126,60 +152,50 @@ def from_batch_entry(cls, entry: PublishBatchRequestEntry, is_fifo=False) -> "Sn
         )
 
 
-class SnsSubscription(TypedDict, total=False):
-    """
-    In SNS, Subscription can be represented with only TopicArn, Endpoint, Protocol, SubscriptionArn and Owner, for
-    example in ListSubscriptions. However, when getting a subscription with GetSubscriptionAttributes, it will return
-    the Subscription object merged with its own attributes.
-    This represents this merged object, for internal use and in GetSubscriptionAttributes
-    https://docs.aws.amazon.com/cli/latest/reference/sns/get-subscription-attributes.html
-    """
+@dataclass
+class PlatformEndpoint:
+    platform_application_arn: str
+    platform_endpoint: Endpoint
 
-    TopicArn: topicARN
-    Endpoint: str
-    Protocol: SnsProtocols
-    SubscriptionArn: subscriptionARN
-    PendingConfirmation: Literal["true", "false"]
-    Owner: str | None
-    SubscriptionPrincipal: str | None
-    FilterPolicy: str | None
-    FilterPolicyScope: Literal["MessageAttributes", "MessageBody"]
-    RawMessageDelivery: Literal["true", "false"]
-    ConfirmationWasAuthenticated: Literal["true", "false"]
-    SubscriptionRoleArn: str | None
-    DeliveryPolicy: str | None
+
+@dataclass
+class PlatformApplicationDetails:
+    platform_application: PlatformApplication
+    # maps all Endpoints of the PlatformApplication, from their Token to their ARN
+    platform_endpoints: dict[str, str]
 
 
 class SnsStore(BaseStore):
-    # maps topic ARN to subscriptions ARN
-    topic_subscriptions: dict[str, list[str]] = LocalAttribute(default=dict)
+    # maps topic ARN to Topic
+    topics: dict[str, Topic] = LocalAttribute(default=dict)
 
     # maps subscription ARN to SnsSubscription
     subscriptions: dict[str, SnsSubscription] = LocalAttribute(default=dict)
 
+    # filter policy are stored as JSON string in subscriptions, store the decoded result Dict
+    subscription_filter_policy: dict[subscriptionARN, dict] = LocalAttribute(default=dict)
+
     # maps confirmation token to subscription ARN
     subscription_tokens: dict[str, str] = LocalAttribute(default=dict)
 
-    # maps topic ARN to list of tags
-    sns_tags: dict[str, list[dict]] = LocalAttribute(default=dict)
+    # maps platform application arns to platform applications
+    platform_applications: dict[str, PlatformApplicationDetails] = LocalAttribute(default=dict)
+
+    # maps endpoint arns to platform endpoints
+    platform_endpoints: dict[str, PlatformEndpoint] = LocalAttribute(default=dict)
 
     # cache of topic ARN to platform endpoint messages (used primarily for testing)
     platform_endpoint_messages: dict[str, list[dict]] = LocalAttribute(default=dict)
 
+    # topic/subscription independent default values for sending sms messages
+    sms_attributes: dict[str, str] = LocalAttribute(default=dict)
+
     # list of sent SMS messages
     sms_messages: list[dict] = LocalAttribute(default=list)
 
-    # filter policy are stored as JSON string in subscriptions, store the decoded result Dict
-    subscription_filter_policy: dict[subscriptionARN, dict] = LocalAttribute(default=dict)
+    TAGS: TaggingService = CrossRegionAttribute(default=TaggingService)
 
-    def get_topic_subscriptions(self, topic_arn: str) -> list[SnsSubscription]:
-        topic_subscriptions = self.topic_subscriptions.get(topic_arn, [])
-        subscriptions = [
-            subscription
-            for subscription_arn in topic_subscriptions
-            if (subscription := self.subscriptions.get(subscription_arn))
-        ]
-        return subscriptions
+    PHONE_NUMBERS_OPTED_OUT: set[PhoneNumber] = CrossRegionAttribute(default=set)
 
 
 sns_stores = AccountRegionBundle("sns", SnsStore)