AWS AWSSimpleQueueService documentation change
Summary
Added a comprehensive Python code example demonstrating SNS-SQS integration with FIFO topics, message filtering, deduplication, and resource cleanup
Security assessment
The change adds documentation about security best practices including: 1) Resource-based queue policies restricting message reception to specific SNS topics (Condition: aws:SourceArn), 2) Guidance on FIFO deduplication to prevent duplicate message processing, 3) Secure cleanup procedures for subscriptions and resources. However, it doesn't address any specific vulnerability or incident.
Diff
diff --git a/AWSSimpleQueueService/latest/SQSDeveloperGuide/example_sqs_Scenario_TopicsAndQueues_section.md b/AWSSimpleQueueService/latest/SQSDeveloperGuide/example_sqs_Scenario_TopicsAndQueues_section.md index 9bd397438..f8d0edb9c 100644 --- a//AWSSimpleQueueService/latest/SQSDeveloperGuide/example_sqs_Scenario_TopicsAndQueues_section.md +++ b//AWSSimpleQueueService/latest/SQSDeveloperGuide/example_sqs_Scenario_TopicsAndQueues_section.md @@ -4066,0 +4067,906 @@ There's more on GitHub. Find the complete example and learn how to set up and ru +Python + + +**SDK for Python (Boto3)** + + +###### Note + +There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/topics_and_queues#code-examples). + +Run an interactive scenario at a command prompt. + + + class TopicsAndQueuesScenario: + """Manages the Topics and Queues feature scenario.""" + + DASHES = "-" * 80 + + def __init__(self, sns_wrapper: SnsWrapper, sqs_wrapper: SqsWrapper) -> None: + """ + Initialize the Topics and Queues scenario. + + :param sns_wrapper: SnsWrapper instance for SNS operations. + :param sqs_wrapper: SqsWrapper instance for SQS operations. + """ + self.sns_wrapper = sns_wrapper + self.sqs_wrapper = sqs_wrapper + + # Scenario state + self.use_fifo_topic = False + self.use_content_based_deduplication = False + self.topic_name = None + self.topic_arn = None + self.queue_count = 2 + self.queue_urls = [] + self.subscription_arns = [] + self.tones = ["cheerful", "funny", "serious", "sincere"] + + def run_scenario(self) -> None: + """Run the Topics and Queues feature scenario.""" + print(self.DASHES) + print("Welcome to messaging with topics and queues.") + print(self.DASHES) + print(f""" + In this scenario, you will create an SNS topic and subscribe {self.queue_count} SQS queues to the topic. + You can select from several options for configuring the topic and the subscriptions for the queues. + You can then post to the topic and see the results in the queues. + """) + + try: + # Setup Phase + print(self.DASHES) + self._setup_topic() + print(self.DASHES) + + self._setup_queues() + print(self.DASHES) + + # Demonstration Phase + self._publish_messages() + print(self.DASHES) + + # Examination Phase + self._poll_queues_for_messages() + print(self.DASHES) + + # Cleanup Phase + self._cleanup_resources() + print(self.DASHES) + + except Exception as e: + logger.error(f"Scenario failed: {e}") + print(f"There was a problem with the scenario: {e}") + print("\nInitiating cleanup...") + try: + self._cleanup_resources() + except Exception as cleanup_error: + logger.error(f"Error during cleanup: {cleanup_error}") + + print("Messaging with topics and queues scenario is complete.") + print(self.DASHES) + + def _setup_topic(self) -> None: + """Set up the SNS topic to be used with the queues.""" + print("SNS topics can be configured as FIFO (First-In-First-Out).") + print("FIFO topics deliver messages in order and support deduplication and message filtering.") + print() + + self.use_fifo_topic = q.ask("Would you like to work with FIFO topics? (y/n): ", q.is_yesno) + + if self.use_fifo_topic: + print(self.DASHES) + self.topic_name = q.ask("Enter a name for your SNS topic: ", q.non_empty) + print("Because you have selected a FIFO topic, '.fifo' must be appended to the topic name.") + print() + + print(self.DASHES) + print(""" + Because you have chosen a FIFO topic, deduplication is supported. + Deduplication IDs are either set in the message or automatically generated + from content using a hash function. + + If a message is successfully published to an SNS FIFO topic, any message + published and determined to have the same deduplication ID, + within the five-minute deduplication interval, is accepted but not delivered. + + For more information about deduplication, + see https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html. + """) + + self.use_content_based_deduplication = q.ask( + "Use content-based deduplication instead of entering a deduplication ID? (y/n): ", + q.is_yesno + ) + else: + self.topic_name = q.ask("Enter a name for your SNS topic: ", q.non_empty) + + print(self.DASHES) + + # Create the topic + self.topic_arn = self.sns_wrapper.create_topic( + self.topic_name, + self.use_fifo_topic, + self.use_content_based_deduplication + ) + + print(f"Your new topic with the name {self.topic_name}") + print(f" and Amazon Resource Name (ARN) {self.topic_arn}") + print(f" has been created.") + print() + + def _setup_queues(self) -> None: + """Set up the SQS queues and subscribe them to the topic.""" + print(f"Now you will create {self.queue_count} Amazon Simple Queue Service (Amazon SQS) queues to subscribe to the topic.") + + for i in range(self.queue_count): + queue_name = q.ask(f"Enter a name for SQS queue #{i+1}: ", q.non_empty) + + if self.use_fifo_topic and i == 0: + print("Because you have selected a FIFO topic, '.fifo' must be appended to the queue name.") + + # Create the queue + queue_url = self.sqs_wrapper.create_queue(queue_name, self.use_fifo_topic) + self.queue_urls.append(queue_url) + + print(f"Your new queue with the name {queue_name}") + print(f" and queue URL {queue_url}") + print(f" has been created.") + print() + + if i == 0: + print("The queue URL is used to retrieve the queue ARN,") + print("which is used to create a subscription.") + print(self.DASHES) + + # Get queue ARN + queue_arn = self.sqs_wrapper.get_queue_arn(queue_url) + + if i == 0: + print("An AWS Identity and Access Management (IAM) policy must be attached to an SQS queue,") + print("enabling it to receive messages from an SNS topic.") + + # Set queue policy to allow SNS to send messages + self.sqs_wrapper.set_queue_policy_for_topic(queue_arn, self.topic_arn, queue_url) + + # Set up message filtering if using FIFO + subscription_arn = self._setup_subscription_with_filter(i, queue_arn, queue_name) + self.subscription_arns.append(subscription_arn) + + def _setup_subscription_with_filter(self, queue_index: int, queue_arn: str, queue_name: str) -> str: + """Set up subscription with optional message filtering.""" + filter_policy = None + + if self.use_fifo_topic: + print(self.DASHES) + if queue_index == 0: + print("Subscriptions to a FIFO topic can have filters.") + print("If you add a filter to this subscription, then only the filtered messages") + print("will be received in the queue.") + print() + print("For information about message filtering,") + print("see https://docs.aws.amazon.com/sns/latest/dg/sns-message-filtering.html") + print() + print("For this example, you can filter messages by a TONE attribute.") + + use_filter = q.ask(f"Filter messages for {queue_name}'s subscription to the topic? (y/n): ", q.is_yesno) + + if use_filter: + filter_policy = self._create_filter_policy() + + subscription_arn = self.sns_wrapper.subscribe_queue_to_topic( + self.topic_arn, queue_arn, filter_policy + ) + + print(f"The queue {queue_name} has been subscribed to the topic {self.topic_name}")