SNS: Disable cross-region access (#6355)
This commit is contained in:
parent
37cb6cee94
commit
c171781223
@ -493,7 +493,7 @@ class SNSBackend(BaseBackend):
|
|||||||
def get_topic(self, arn: str) -> Topic:
|
def get_topic(self, arn: str) -> Topic:
|
||||||
parsed_arn = parse_arn(arn)
|
parsed_arn = parse_arn(arn)
|
||||||
try:
|
try:
|
||||||
return sns_backends[parsed_arn.account][parsed_arn.region].topics[arn]
|
return sns_backends[parsed_arn.account][self.region_name].topics[arn]
|
||||||
except KeyError:
|
except KeyError:
|
||||||
raise TopicNotFound
|
raise TopicNotFound
|
||||||
|
|
||||||
|
@ -1,7 +1,8 @@
|
|||||||
import boto3
|
import boto3
|
||||||
import json
|
import json
|
||||||
|
import pytest
|
||||||
|
|
||||||
# import sure # noqa # pylint: disable=unused-import
|
import sure # noqa # pylint: disable=unused-import
|
||||||
|
|
||||||
from botocore.exceptions import ClientError
|
from botocore.exceptions import ClientError
|
||||||
from moto import mock_sns
|
from moto import mock_sns
|
||||||
@ -131,9 +132,23 @@ def test_create_topic_should_be_of_certain_length():
|
|||||||
def test_create_topic_in_multiple_regions():
|
def test_create_topic_in_multiple_regions():
|
||||||
for region in ["us-west-1", "us-west-2"]:
|
for region in ["us-west-1", "us-west-2"]:
|
||||||
conn = boto3.client("sns", region_name=region)
|
conn = boto3.client("sns", region_name=region)
|
||||||
conn.create_topic(Name="some-topic")
|
topic_arn = conn.create_topic(Name="some-topic")["TopicArn"]
|
||||||
|
# We can find the topic
|
||||||
list(conn.list_topics()["Topics"]).should.have.length_of(1)
|
list(conn.list_topics()["Topics"]).should.have.length_of(1)
|
||||||
|
|
||||||
|
# We can read the Topic details
|
||||||
|
topic = boto3.resource("sns", region_name=region).Topic(topic_arn)
|
||||||
|
topic.load()
|
||||||
|
|
||||||
|
# Topic does not exist in different region though
|
||||||
|
with pytest.raises(ClientError) as exc:
|
||||||
|
sns_resource = boto3.resource("sns", region_name="eu-north-1")
|
||||||
|
topic = sns_resource.Topic(topic_arn)
|
||||||
|
topic.load()
|
||||||
|
err = exc.value.response["Error"]
|
||||||
|
assert err["Code"] == "NotFound"
|
||||||
|
assert err["Message"] == "Topic does not exist"
|
||||||
|
|
||||||
|
|
||||||
@mock_sns
|
@mock_sns
|
||||||
def test_topic_corresponds_to_region():
|
def test_topic_corresponds_to_region():
|
||||||
|
Loading…
Reference in New Issue
Block a user