AWS code-library documentation change
Summary
Added comprehensive Python code example demonstrating an interactive scenario for using Amazon SNS and Amazon SQS together, including FIFO topics/queues, message deduplication, filtering, and resource cleanup.
Security assessment
The change adds documentation about security features through the implementation of SQS resource policies that restrict message reception to specific SNS topics (using aws:SourceArn condition). This demonstrates the security best practice of least-privilege access. However, there's no evidence this addresses a specific security vulnerability.
Diff
diff --git a/code-library/latest/ug/sqs_example_sqs_Scenario_TopicsAndQueues_section.md b/code-library/latest/ug/sqs_example_sqs_Scenario_TopicsAndQueues_section.md index 6aa0b2e5c..c1d138b9e 100644 --- a//code-library/latest/ug/sqs_example_sqs_Scenario_TopicsAndQueues_section.md +++ b//code-library/latest/ug/sqs_example_sqs_Scenario_TopicsAndQueues_section.md @@ -4068,0 +4069,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}")