From c7e38afeb1dd1e934a96fea340addf08eec85567 Mon Sep 17 00:00:00 2001 From: Casey Bodley Date: Tue, 12 Mar 2024 20:05:15 -0400 Subject: [PATCH] sns: add test_sns.py for simple topic testing Signed-off-by: Casey Bodley (cherry picked from commit dfabbf5a8d835424147338693b4033c708b24543) --- pytest.ini | 2 +- s3tests_boto3/functional/test_sns.py | 159 +++++++++++++++++++++++++++ 2 files changed, 160 insertions(+), 1 deletion(-) create mode 100644 s3tests_boto3/functional/test_sns.py diff --git a/pytest.ini b/pytest.ini index c7e0418..73d1563 100644 --- a/pytest.ini +++ b/pytest.ini @@ -34,7 +34,7 @@ markers = s3website s3website_routing_rules s3website_redirect_location - 3website + sns sse_s3 storage_class tagging diff --git a/s3tests_boto3/functional/test_sns.py b/s3tests_boto3/functional/test_sns.py new file mode 100644 index 0000000..360f14e --- /dev/null +++ b/s3tests_boto3/functional/test_sns.py @@ -0,0 +1,159 @@ +import json +import pytest +from botocore.exceptions import ClientError +from . import ( + configfile, + get_iam_root_client, + get_iam_alt_root_client, + get_new_bucket_name, + get_prefix, + nuke_prefixed_buckets, +) +from .iam import iam_root, iam_alt_root +from .utils import assert_raises, _get_status_and_error_code + +def get_new_topic_name(): + return get_new_bucket_name() + +def nuke_topics(client, prefix): + p = client.get_paginator('list_topics') + for response in p.paginate(): + for topic in response['Topics']: + arn = topic['TopicArn'] + if prefix not in arn: + pass + try: + client.delete_topic(TopicArn=arn) + except: + pass + +@pytest.fixture +def sns(iam_root): + client = get_iam_root_client(service_name='sns') + yield client + nuke_topics(client, get_prefix()) + +@pytest.fixture +def sns_alt(iam_alt_root): + client = get_iam_alt_root_client(service_name='sns') + yield client + nuke_topics(client, get_prefix()) + +@pytest.fixture +def s3(iam_root): + client = get_iam_root_client(service_name='s3') + yield client + nuke_prefixed_buckets(get_prefix(), client) + +@pytest.fixture +def s3_alt(iam_alt_root): + client = get_iam_alt_root_client(service_name='s3') + yield client + nuke_prefixed_buckets(get_prefix(), client) + + +@pytest.mark.iam_account +@pytest.mark.sns +def test_account_topic(sns): + name = get_new_topic_name() + + response = sns.create_topic(Name=name) + arn = response['TopicArn'] + assert arn.startswith('arn:aws:sns:') + assert arn.endswith(f':{name}') + + response = sns.list_topics() + assert arn in [p['TopicArn'] for p in response['Topics']] + + sns.set_topic_attributes(TopicArn=arn, AttributeName='Policy', AttributeValue='') + + response = sns.get_topic_attributes(TopicArn=arn) + assert 'Attributes' in response + + sns.delete_topic(TopicArn=arn) + + response = sns.list_topics() + assert arn not in [p['TopicArn'] for p in response['Topics']] + + with pytest.raises(sns.exceptions.NotFoundException): + sns.get_topic_attributes(TopicArn=arn) + sns.delete_topic(TopicArn=arn) + +@pytest.mark.iam_account +@pytest.mark.sns +def test_cross_account_topic(sns, sns_alt): + name = get_new_topic_name() + arn = sns.create_topic(Name=name)['TopicArn'] + + # not visible to any alt user apis + with pytest.raises(sns.exceptions.NotFoundException): + sns_alt.get_topic_attributes(TopicArn=arn) + with pytest.raises(sns.exceptions.NotFoundException): + sns_alt.set_topic_attributes(TopicArn=arn, AttributeName='Policy', AttributeValue='') + + # delete returns success + sns_alt.delete_topic(TopicArn=arn) + + response = sns_alt.list_topics() + assert arn not in [p['TopicArn'] for p in response['Topics']] + +@pytest.mark.iam_account +@pytest.mark.sns +def test_account_topic_publish(sns, s3): + name = get_new_topic_name() + + response = sns.create_topic(Name=name) + topic_arn = response['TopicArn'] + + bucket = get_new_bucket_name() + s3.create_bucket(Bucket=bucket) + + config = {'TopicConfigurations': [{ + 'Id': 'id', + 'TopicArn': topic_arn, + 'Events': [ 's3:ObjectCreated:*' ], + }]} + s3.put_bucket_notification_configuration( + Bucket=bucket, NotificationConfiguration=config) + +@pytest.mark.iam_account +@pytest.mark.iam_cross_account +@pytest.mark.sns +def test_cross_account_topic_publish(sns, s3_alt, iam_alt_root): + name = get_new_topic_name() + + response = sns.create_topic(Name=name) + topic_arn = response['TopicArn'] + + bucket = get_new_bucket_name() + s3_alt.create_bucket(Bucket=bucket) + + config = {'TopicConfigurations': [{ + 'Id': 'id', + 'TopicArn': topic_arn, + 'Events': [ 's3:ObjectCreated:*' ], + }]} + + # expect AccessDenies because no resource policy allows cross-account access + e = assert_raises(ClientError, s3_alt.put_bucket_notification_configuration, + Bucket=bucket, NotificationConfiguration=config) + status, error_code = _get_status_and_error_code(e.response) + assert status == 403 + assert error_code == 'AccessDenied' + + # add topic policy to allow the alt user + alt_principal = iam_alt_root.get_user()['User']['Arn'] + policy = json.dumps({ + 'Version': '2012-10-17', + 'Statement': [{ + 'Effect': 'Allow', + 'Principal': {'AWS': alt_principal}, + 'Action': 'sns:Publish', + 'Resource': topic_arn + }] + }) + sns.set_topic_attributes(TopicArn=topic_arn, AttributeName='Policy', + AttributeValue=policy) + + s3_alt.put_bucket_notification_configuration( + Bucket=bucket, NotificationConfiguration=config)