AWS：通过 boto3 订阅 SNS 时，SQS 队列为空

Question

通过 boto3 订阅 SNS 主题时，我没有在 SQS 队列中收到任何消息。

这是我使用的代码或 API 凭据的问题吗？与此账户关联的 IAM 策略具有 AWS PowerUser 权限，这意味着它可以不受限制地访问管理 SNS 主题和 SQS 队列。

当我通过 AWS 控制台创建等效结构（创建主题、创建队列、订阅主题队列）并使用 boto3、AWS CLI 或 AWS 控制台发送消息时，消息正确传递。

我不认为这是代码的问题，因为SubscriptionArn正确返回了？

我在 US-EAST-1 和 AP-SE-1 区域都尝试过这个，结果相同。

示例代码：

#!/usr/bin/env python3

import boto3
import json

def get_sqs_msgs_from_sns():
    sqs_client = boto3.client('sqs', region_name='us-east-1')
    sqs_obj = boto3.resource('sqs', region_name='us-east-1')
    sns_client = boto3.client('sns', region_name='us-east-1')
    sqs_queue_name = 'queue1'
    topic_name = 'topic1'

    # Create/Get Queue
    sqs_client.create_queue(QueueName=sqs_queue_name)
    sqs_queue = sqs_obj.get_queue_by_name(QueueName=sqs_queue_name)
    queue_url = sqs_client.get_queue_url(QueueName=sqs_queue_name)['QueueUrl']
    sqs_queue_attrs = sqs_client.get_queue_attributes(QueueUrl=queue_url,
                                                    AttributeNames=['All'])['Attributes']
    sqs_queue_arn = sqs_queue_attrs['QueueArn']
    if ':sqs.' in sqs_queue_arn:
        sqs_queue_arn = sqs_queue_arn.replace(':sqs.', ':')

    # Create SNS Topic
    topic_res = sns_client.create_topic(Name=topic_name)
    sns_topic_arn = topic_res['TopicArn']

    # Subscribe SQS queue to SNS
    sns_client.subscribe(
            TopicArn=sns_topic_arn,
            Protocol='sqs',
            Endpoint=sqs_queue_arn
    )

    # Publish SNS Messages
    test_msg = {'default': {"x":"foo","y":"bar"}}
    test_msg_body = json.dumps(test_msg)
    sns_client.publish(
        TopicArn=sns_topic_arn, 
        Message=json.dumps({'default': test_msg_body}),
        MessageStructure='json')

    # Validate Message
    sqs_msgs = sqs_queue.receive_messages(
            AttributeNames=['All'],
            MessageAttributeNames=['All'],
            VisibilityTimeout=15,
            WaitTimeSeconds=20,
            MaxNumberOfMessages=5
    )
    assert len(sqs_msgs) == 1
    assert sqs_msgs[0].body == test_msg_body
    print(sqs_msgs[0].body) # This should output dict with keys Message, Type, Timestamp, etc., but only returns the test_msg

if __name__ == "__main__":
    get_mock_sqs_msgs_from_sns()

我收到这个输出：

$ python .\sns-test.py
Traceback (most recent call last):
  File ".\sns-test.py", line 55, in <module>
    get_sqs_msgs_from_sns()
  File ".\sns-test.py", line 50, in get_sqs_msgs_from_sns
    assert len(sqs_msgs) == 1
AssertionError

Answer 1

上面针对 C# AWS SDK 提出的类似问题的 URL 使我朝着正确的方向前进：我需要将策略附加到 SQS 队列以允许 SNS 主题写入它。

def allow_sns_to_write_to_sqs(topicarn, queuearn):
    policy_document = """{{
  "Version":"2012-10-17",
  "Statement":[
    {{
      "Sid":"MyPolicy",
      "Effect":"Allow",
      "Principal" : {{"AWS" : "*"}},
      "Action":"SQS:SendMessage",
      "Resource": "{}",
      "Condition":{{
        "ArnEquals":{{
          "aws:SourceArn": "{}"
        }}
      }}
    }}
  ]
}}""".format(queuearn, topicarn)

    return policy_document

和

policy_json = allow_sns_to_write_to_sqs(topic_arn, queue_arn)

response = sqs_client.set_queue_attributes(
    QueueUrl = queue_url,
    Attributes = {
        'Policy' : policy_json
    }
)
print(response)