Back to Checks
aws

SNS topic is not publicly accessible

sns_topics_not_publicly_accessible
Service: snsSeverity: high
OfficialBuilt-inVerified
Categories:internet-exposed
Resource Type:AwsSnsTopic

SNS topic policies are analyzed for public principals (e.g., *). Topics that grant access without restrictive conditions such as aws:SourceArn, aws:SourceAccount, aws:PrincipalOrgID, or sns:Endpoint scoping are treated as publicly accessible.

Risk

Public SNS topics allow anyone or unknown accounts to:

  • Subscribe and siphon messages (confidentiality)
  • Publish spoofed payloads that alter workflows (integrity)
  • Flood messages causing outages and costs (availability) They also enable cross-account abuse and bypass expected trust boundaries.

Run this check with Prowler

Run in Prowler Cloud
Run
prowler aws --checks sns_topics_not_publicly_accessible

Recommendation

Restrict the topic policy to specific principals and minimal actions:

  • Avoid Principal:*
  • Allow only needed actions (e.g., sns:Publish)
  • Add conditions like aws:SourceArn, aws:SourceAccount, aws:PrincipalOrgID, or sns:Endpoint Apply least privilege, separate duties, and review policies regularly.

Remediation

CLI

aws sns set-topic-attributes --topic-arn <TOPIC_ARN> --attribute-name Policy --attribute-value '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":"arn:aws:iam::<ACCOUNT_ID>:root"},"Action":"sns:Publish","Resource":"<TOPIC_ARN>"}]}'

Terraform

# Restrict SNS topic policy to the account (not public)
resource "aws_sns_topic_policy" "<example_resource_name>" {
  arn    = "<TOPIC_ARN>"
  policy = jsonencode({
    Version   = "2012-10-17"
    Statement = [{
      Effect    = "Allow"
      Action    = "sns:Publish"
      Resource  = "<TOPIC_ARN>"
      Principal = { AWS = "arn:aws:iam::<ACCOUNT_ID>:root" } # Critical: restrict principal to the account to remove public access
    }]
  })
}

NativeIaC

# CloudFormation: restrict SNS topic policy to the account (not public)
Resources:
  <example_resource_name>:
    Type: AWS::SNS::TopicPolicy
    Properties:
      Topics:
        - arn:aws:sns:<region>:<account_id>:<example_resource_name>
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Action: sns:Publish
            Resource: arn:aws:sns:<region>:<account_id>:<example_resource_name>
            Principal:
              AWS: arn:aws:iam::<account_id>:root  # Critical: restrict to account root to remove public access

Other

1Open the Amazon SNS console and select Topics
2Choose the topic and go to the Access policy tab
3Edit the policy and remove any Principal set to "*" (Everyone/Public)
4Add a statement allowing only your account root: Principal = arn:aws:iam::<ACCOUNT_ID>:root with Action sns:Publish and Resource set to the topic ARN
5Save changes

Check Code

from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.iam.lib.policy import (
    has_public_principal,
    has_restrictive_source_arn_condition,
    is_condition_block_restrictive,
    is_condition_block_restrictive_organization,
    is_condition_block_restrictive_sns_endpoint,
)
from prowler.providers.aws.services.sns.sns_client import sns_client


class sns_topics_not_publicly_accessible(Check):
    def execute(self):
        findings = []
        for topic in sns_client.topics:
            report = Check_Report_AWS(metadata=self.metadata(), resource=topic)
            report.status = "PASS"
            report.status_extended = (
                f"SNS topic {topic.name} is not publicly accessible."
            )

            if topic.policy:
                for statement in topic.policy["Statement"]:
                    if statement["Effect"] == "Allow" and has_public_principal(
                        statement
                    ):
                        if has_restrictive_source_arn_condition(statement):
                            break
                        elif "Condition" in statement:
                            condition_account = is_condition_block_restrictive(
                                statement["Condition"], sns_client.audited_account
                            )
                            condition_org = is_condition_block_restrictive_organization(
                                statement["Condition"]
                            )
                            condition_endpoint = (
                                is_condition_block_restrictive_sns_endpoint(
                                    statement["Condition"]
                                )
                            )

                            if condition_account and condition_org:
                                report.status_extended = f"SNS topic {topic.name} is not public because its policy only allows access from the account {sns_client.audited_account} and an organization."
                            elif condition_account:
                                report.status_extended = f"SNS topic {topic.name} is not public because its policy only allows access from the account {sns_client.audited_account}."
                            elif condition_org:
                                report.status_extended = f"SNS topic {topic.name} is not public because its policy only allows access from an organization."
                            elif condition_endpoint:
                                report.status_extended = f"SNS topic {topic.name} is not public because its policy only allows access from an endpoint."
                            else:
                                report.status = "FAIL"
                                report.status_extended = f"SNS topic {topic.name} is public because its policy allows public access."
                                break
                        else:
                            # Public principal with no conditions = public
                            report.status = "FAIL"
                            report.status_extended = f"SNS topic {topic.name} is public because its policy allows public access."
                            break

            findings.append(report)
        return findings