Quishu
Reputation: 1
Subscribing lambda to SNS using boto3. Lambda trigger is not set
I have created script that creates SNS topic and subscribe already created lambda to it. Everything works fine, except that Lambda trigger is not set to SNS topic. If I try to publish message to the topic I get in a CloudWatch Logs message with this information:
{\"ErrorCode\":\"AccessDeniedException\",\"ErrorMessage\":\"User: sns.amazonaws.com is not authorized to perform: lambda:InvokeFunction on resource: arn:aws:lambda:us-east-1:<my_account_id_here>:function:sns-test-lambda because no resource-based policy allows the lambda:InvokeFunction action\",\"lambdaRequestId\":\"Unrecoverable\"}
If the problem is that: sns.amazonaws.com is not authorized to perform: lambda:InvokeFunction, how can I set policy to SNS that allows to invoke lambda function?
I tried to create topic programatically without subscribing and then subscribe using AWS console and it worked- lambda was a subscriber on SNS topic panel and SNS topic was indicated in Lambda triggers panel, which is what I'm looking for to achieve (but programatically). When I publish a message in that scenario everyting works fine, lambda is executing, CloudWatch Logs appears with succesful status.
Python script:
import boto3
import json
import logging
from botocore.exceptions import ClientError
logger = logging.getLogger(__name__)
boto3.setup_default_session(profile_name='xxx')
class SnsWrapper:
"""Encapsulates Amazon SNS topic and subscription functions."""
def __init__(self, sns_resource):
"""
:param sns_resource: A Boto3 Amazon SNS resource.
"""
self.sns_resource = sns_resource
def create_topic(self, name, attributes=None):
"""
Creates a notification topic.
:param name: The name of the topic to create.
:param attributes: Attributes of topic.
:return: The newly created topic.
"""
try:
if attributes:
topic = self.sns_resource.create_topic(Name=name, Attributes=attributes)
else:
topic = self.sns_resource.create_topic(Name=name)
logger.info("Created topic %s with ARN %s.", name, topic.arn)
except ClientError:
logger.exception("Couldn't create topic %s.", name)
raise
else:
return topic
def list_topics(self):
"""
Lists topics for the current account.
:return: An iterator that yields the topics.
"""
try:
topics_iter = self.sns_resource.topics.all()
logger.info("Got topics.")
except ClientError:
logger.exception("Couldn't get topics.")
raise
else:
return topics_iter
@staticmethod
def delete_topic(topic):
"""
Deletes a topic. All subscriptions to the topic are also deleted.
"""
try:
topic.delete()
logger.info("Deleted topic %s.", topic.arn)
except ClientError:
logger.exception("Couldn't delete topic %s.", topic.arn)
raise
@staticmethod
def subscribe(topic, protocol, endpoint):
"""
Subscribes an endpoint to the topic. Some endpoint types, such as email,
must be confirmed before their subscriptions are active. When a subscription
is not confirmed, its Amazon Resource Number (ARN) is set to
'PendingConfirmation'.
:param topic: The topic to subscribe to.
:param protocol: The protocol of the endpoint, such as 'sms' or 'email'.
:param endpoint: The endpoint that receives messages, such as a phone number
(in E.164 format) for SMS messages, or an email address for
email messages.
:return: The newly added subscription.
"""
try:
subscription = topic.subscribe(
Protocol=protocol, Endpoint=endpoint, ReturnSubscriptionArn=True)
logger.info("Subscribed %s %s to topic %s.", protocol, endpoint, topic.arn)
except ClientError:
logger.exception(
"Couldn't subscribe %s %s to topic %s.", protocol, endpoint, topic.arn)
raise
else:
return subscription
def list_subscriptions(self, topic=None):
"""
Lists subscriptions for the current account, optionally limited to a
specific topic.
:param topic: When specified, only subscriptions to this topic are returned.
:return: An iterator that yields the subscriptions.
"""
try:
if topic is None:
subs_iter = self.sns_resource.subscriptions.all()
else:
subs_iter = topic.subscriptions.all()
logger.info("Got subscriptions.")
except ClientError:
logger.exception("Couldn't get subscriptions.")
raise
else:
return subs_iter
@staticmethod
def add_subscription_filter(subscription, attributes):
"""
Adds a filter policy to a subscription. A filter policy is a key and a
list of values that are allowed. When a message is published, it must have an
attribute that passes the filter or it will not be sent to the subscription.
:param subscription: The subscription the filter policy is attached to.
:param attributes: A dictionary of key-value pairs that define the filter.
"""
try:
att_policy = {key: [value] for key, value in attributes.items()}
subscription.set_attributes(
AttributeName='FilterPolicy', AttributeValue=json.dumps(att_policy))
logger.info("Added filter to subscription %s.", subscription.arn)
except ClientError:
logger.exception(
"Couldn't add filter to subscription %s.", subscription.arn)
raise
@staticmethod
def delete_subscription(subscription):
"""
Unsubscribes and deletes a subscription.
"""
try:
subscription.delete()
logger.info("Deleted subscription %s.", subscription.arn)
except ClientError:
logger.exception("Couldn't delete subscription %s.", subscription.arn)
raise
def publish_text_message(self, phone_number, message):
"""
Publishes a text message directly to a phone number without need for a
subscription.
:param phone_number: The phone number that receives the message. This must be
in E.164 format. For example, a United States phone
number might be +12065550101.
:param message: The message to send.
:return: The ID of the message.
"""
try:
response = self.sns_resource.meta.client.publish(
PhoneNumber=phone_number, Message=message)
message_id = response['MessageId']
logger.info("Published message to %s.", phone_number)
except ClientError:
logger.exception("Couldn't publish message to %s.", phone_number)
raise
else:
return message_id
@staticmethod
def publish_message(topic, message, attributes):
"""
Publishes a message, with attributes, to a topic. Subscriptions can be filtered
based on message attributes so that a subscription receives messages only
when specified attributes are present.
:param topic: The topic to publish to.
:param message: The message to publish.
:param attributes: The key-value attributes to attach to the message. Values
must be either `str` or `bytes`.
:return: The ID of the message.
"""
try:
att_dict = {}
for key, value in attributes.items():
if isinstance(value, str):
att_dict[key] = {'DataType': 'String', 'StringValue': value}
elif isinstance(value, bytes):
att_dict[key] = {'DataType': 'Binary', 'BinaryValue': value}
response = topic.publish(Message=message, MessageAttributes=att_dict)
message_id = response['MessageId']
logger.info(
"Published message with attributes %s to topic %s.", attributes,
topic.arn)
except ClientError:
logger.exception("Couldn't publish message to topic %s.", topic.arn)
raise
else:
return message_id
@staticmethod
def publish_multi_message(
topic, subject, default_message, sms_message, email_message):
"""
Publishes a multi-format message to a topic. A multi-format message takes
different forms based on the protocol of the subscriber. For example,
an SMS subscriber might receive a short, text-only version of the message
while an email subscriber could receive an HTML version of the message.
:param topic: The topic to publish to.
:param subject: The subject of the message.
:param default_message: The default version of the message. This version is
sent to subscribers that have protocols that are not
otherwise specified in the structured message.
:param sms_message: The version of the message sent to SMS subscribers.
:param email_message: The version of the message sent to email subscribers.
:return: The ID of the message.
"""
try:
message = {
'default': default_message,
'sms': sms_message,
'email': email_message
}
response = topic.publish(
Message=json.dumps(message), Subject=subject, MessageStructure='json')
message_id = response['MessageId']
logger.info("Published multi-format message to topic %s.", topic.arn)
except ClientError:
logger.exception("Couldn't publish message to topic %s.", topic.arn)
raise
else:
return message_id
# -----TEST SNS-----
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
sns = boto3.resource('sns')
sns_wrapper = SnsWrapper(sns)
# Setting up topic attributes
retry_policy = str(json.dumps({
"http": {
"defaultHealthyRetryPolicy": {
"minDelayTarget": 20,
"maxDelayTarget": 21,
"numRetries": 3,
"numMaxDelayRetries": 0,
"numNoDelayRetries": 0,
"numMinDelayRetries": 0,
"backoffFunction": "linear"
},
"disableSubscriptionOverrides": False
}
}))
example_access_policy = str(json.dumps({
"Version": "2008-10-17",
"Id": "__default_policy_ID",
"Statement": [
{
"Sid": "__default_statement_ID",
"Effect": "Allow",
"Principal": {
"AWS": "*"
},
"Action": [
"SNS:Publish",
"SNS:RemovePermission",
"SNS:SetTopicAttributes",
"SNS:DeleteTopic",
"SNS:ListSubscriptionsByTopic",
"SNS:GetTopicAttributes",
"SNS:AddPermission",
"SNS:Subscribe"
],
"Resource": "",
"Condition": {
"StringEquals": {
"AWS:SourceOwner": "504215336107"
}
}
}
]
}))
attributes = {
'DeliveryPolicy': retry_policy,
'Policy': example_access_policy,
'KmsMasterKeyId': 'arn:aws:kms:us-east-1:504215336107:key/a0a18770-6a62-418c-b35f-0adee2616270'
}
# Create topic using wrapper
topic = sns_wrapper.create_topic(name='test-sns-functionality', attributes=attributes)
print(topic)
# Lambda ARN (to be subscriber)
lambda_arn = 'arn:aws:lambda:us-east-1:504211336307:function:sns-test-lambda'
# Subscribe lambda to created topic
subscriber_arn = sns_wrapper.subscribe(topic, 'lambda', lambda_arn)
# Publish message to topic
sns_wrapper.publish_message(topic, f'This is message from topic: {topic}', attributes={'atr1': 'val1'})
Possible solutions: Boto3 docs says:
if the endpoint and the topic are not in the same Amazon Web Services account, the endpoint owner must run the ConfirmSubscription action to confirm the subscription
Not sure how to check who is endpoint owner (by that I mean lambda owner), but I have created lambda and SNS topic using the same account.
Does anyone know what am I doing wrong or where the problem is?
Upvotes: 0
Views: 1143
Answers (1)
Quishu
Reputation: 1
Issue:
The problem was that sns.amazonaws.com is not authorized to perform: lambda:InvokeFunction. I am still not 100% sure, if there is any way to set up SNS that invokes Lambda without interfering with lambda itself using boto3.
My solution:
After setting up Resource-based policy in Lambda like this:
Statement ID: <name-of-your-policy-statement>
Principal: sns.amazonaws.com
Effect: Allow
Action: lambda:InvokeFunction
Conditions:
{
"ArnLike": {
"AWS:SourceArn": "arn:aws:sns:us-east-1:504211136307:test-sns-functionality"
}
}
Lambda triggers were populated automatically with SNS Topic that I chose (AWS:SourceArn - is a SNS Topic arn in this example)
Additional information:
Logging messages status is really important, but I haven't found any piece of information about setting up topics Delivery Status Loggings using boto3. Luckily tried to set attributes used in topic creation methods and it worked!
'LambdaSuccessFeedbackRoleArn': 'arn:aws:iam::504255336111:role/SNS_CloudWatch_Logs_Access_Role',
'LambdaFailureFeedbackRoleArn': 'arn:aws:iam::504255336111:role/SNS_CloudWatch_Logs_Access_Role'
Working script that creates SNS topic (with attributes like logging and policies), subscribes lambda, adds permissions to lambda, publishes message to created topic.
import boto3
import json
import logging
from botocore.exceptions import ClientError
logger = logging.getLogger(__name__)
boto3.setup_default_session(profile_name='xxx')
class SnsWrapper:
"""Encapsulates Amazon SNS topic and subscription functions."""
def __init__(self, sns_resource):
"""
:param sns_resource: A Boto3 Amazon SNS resource.
"""
self.sns_resource = sns_resource
def create_topic(self, name, attributes=None):
"""
Creates a notification topic.
:param name: The name of the topic to create.
:param attributes: Attributes of topic.
:return: The newly created topic.
"""
try:
if attributes:
topic = self.sns_resource.create_topic(Name=name, Attributes=attributes)
else:
topic = self.sns_resource.create_topic(Name=name)
logger.info("Created topic %s with ARN %s.", name, topic.arn)
except ClientError:
logger.exception("Couldn't create topic %s.", name)
raise
else:
return topic
def list_topics(self):
"""
Lists topics for the current account.
:return: An iterator that yields the topics.
"""
try:
topics_iter = self.sns_resource.topics.all()
logger.info("Got topics.")
except ClientError:
logger.exception("Couldn't get topics.")
raise
else:
return topics_iter
@staticmethod
def delete_topic(topic):
"""
Deletes a topic. All subscriptions to the topic are also deleted.
"""
try:
topic.delete()
logger.info("Deleted topic %s.", topic.arn)
except ClientError:
logger.exception("Couldn't delete topic %s.", topic.arn)
raise
@staticmethod
def subscribe(topic, protocol, endpoint):
"""
Subscribes an endpoint to the topic. Some endpoint types, such as email,
must be confirmed before their subscriptions are active. When a subscription
is not confirmed, its Amazon Resource Number (ARN) is set to
'PendingConfirmation'.
:param topic: The topic to subscribe to.
:param protocol: The protocol of the endpoint, such as 'sms' or 'email'.
:param endpoint: The endpoint that receives messages, such as a phone number
(in E.164 format) for SMS messages, or an email address for
email messages.
:return: The newly added subscription.
"""
try:
subscription = topic.subscribe(
Protocol=protocol, Endpoint=endpoint, ReturnSubscriptionArn=True)
logger.info("Subscribed %s %s to topic %s.", protocol, endpoint, topic.arn)
except ClientError:
logger.exception(
"Couldn't subscribe %s %s to topic %s.", protocol, endpoint, topic.arn)
raise
else:
return subscription
def list_subscriptions(self, topic=None):
"""
Lists subscriptions for the current account, optionally limited to a
specific topic.
:param topic: When specified, only subscriptions to this topic are returned.
:return: An iterator that yields the subscriptions.
"""
try:
if topic is None:
subs_iter = self.sns_resource.subscriptions.all()
else:
subs_iter = topic.subscriptions.all()
logger.info("Got subscriptions.")
except ClientError:
logger.exception("Couldn't get subscriptions.")
raise
else:
return subs_iter
@staticmethod
def add_subscription_filter(subscription, attributes):
"""
Adds a filter policy to a subscription. A filter policy is a key and a
list of values that are allowed. When a message is published, it must have an
attribute that passes the filter or it will not be sent to the subscription.
:param subscription: The subscription the filter policy is attached to.
:param attributes: A dictionary of key-value pairs that define the filter.
"""
try:
att_policy = {key: [value] for key, value in attributes.items()}
subscription.set_attributes(
AttributeName='FilterPolicy', AttributeValue=json.dumps(att_policy))
logger.info("Added filter to subscription %s.", subscription.arn)
except ClientError:
logger.exception(
"Couldn't add filter to subscription %s.", subscription.arn)
raise
@staticmethod
def delete_subscription(subscription):
"""
Unsubscribes and deletes a subscription.
"""
try:
subscription.delete()
logger.info("Deleted subscription %s.", subscription.arn)
except ClientError:
logger.exception("Couldn't delete subscription %s.", subscription.arn)
raise
def publish_text_message(self, phone_number, message):
"""
Publishes a text message directly to a phone number without need for a
subscription.
:param phone_number: The phone number that receives the message. This must be
in E.164 format. For example, a United States phone
number might be +12065550101.
:param message: The message to send.
:return: The ID of the message.
"""
try:
response = self.sns_resource.meta.client.publish(
PhoneNumber=phone_number, Message=message)
message_id = response['MessageId']
logger.info("Published message to %s.", phone_number)
except ClientError:
logger.exception("Couldn't publish message to %s.", phone_number)
raise
else:
return message_id
@staticmethod
def publish_message(topic, message, attributes):
"""
Publishes a message, with attributes, to a topic. Subscriptions can be filtered
based on message attributes so that a subscription receives messages only
when specified attributes are present.
:param topic: The topic to publish to.
:param message: The message to publish.
:param attributes: The key-value attributes to attach to the message. Values
must be either `str` or `bytes`.
:return: The ID of the message.
"""
try:
att_dict = {}
for key, value in attributes.items():
if isinstance(value, str):
att_dict[key] = {'DataType': 'String', 'StringValue': value}
elif isinstance(value, bytes):
att_dict[key] = {'DataType': 'Binary', 'BinaryValue': value}
response = topic.publish(Message=message, MessageAttributes=att_dict)
message_id = response['MessageId']
logger.info(
"Published message with attributes %s to topic %s.", attributes,
topic.arn)
except ClientError:
logger.exception("Couldn't publish message to topic %s.", topic.arn)
raise
else:
return message_id
@staticmethod
def publish_multi_message(
topic, subject, default_message, sms_message, email_message):
"""
Publishes a multi-format message to a topic. A multi-format message takes
different forms based on the protocol of the subscriber. For example,
an SMS subscriber might receive a short, text-only version of the message
while an email subscriber could receive an HTML version of the message.
:param topic: The topic to publish to.
:param subject: The subject of the message.
:param default_message: The default version of the message. This version is
sent to subscribers that have protocols that are not
otherwise specified in the structured message.
:param sms_message: The version of the message sent to SMS subscribers.
:param email_message: The version of the message sent to email subscribers.
:return: The ID of the message.
"""
try:
message = {
'default': default_message,
'sms': sms_message,
'email': email_message
}
response = topic.publish(
Message=json.dumps(message), Subject=subject, MessageStructure='json')
message_id = response['MessageId']
logger.info("Published multi-format message to topic %s.", topic.arn)
except ClientError:
logger.exception("Couldn't publish message to topic %s.", topic.arn)
raise
else:
return message_id
# -----TEST SNS-----
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
sns = boto3.resource('sns')
lambdas = boto3.client('lambda')
sns_wrapper = SnsWrapper(sns)
# Setting up topic attributes
example_retry_policy = str(json.dumps({
"http": {
"defaultHealthyRetryPolicy": {
"minDelayTarget": 20,
"maxDelayTarget": 21,
"numRetries": 3,
"numMaxDelayRetries": 0,
"numNoDelayRetries": 0,
"numMinDelayRetries": 0,
"backoffFunction": "linear"
},
"disableSubscriptionOverrides": False
}
}))
example_access_policy = str(json.dumps({
"Version": "2008-10-17",
"Id": "__default_policy_ID",
"Statement": [
{
"Sid": "__default_statement_ID",
"Effect": "Allow",
"Principal": {
"AWS": "*"
},
"Action": [
"SNS:Publish",
"SNS:RemovePermission",
"SNS:SetTopicAttributes",
"SNS:DeleteTopic",
"SNS:ListSubscriptionsByTopic",
"SNS:GetTopicAttributes",
"SNS:AddPermission",
"SNS:Subscribe"
],
"Resource": "",
"Condition": {
"StringEquals": {
"AWS:SourceOwner": "504255336111"
}
}
}
]
}))
attributes = {
'DeliveryPolicy': example_retry_policy,
'Policy': example_access_policy,
'KmsMasterKeyId': 'arn:aws:kms:us-east-1:504255336111:key/a0a18770-6a62-468c-b35f-0aded2616111',
'LambdaSuccessFeedbackRoleArn': 'arn:aws:iam::504255336111:role/SNS_CloudWatch_Logs_Access_Role',
'LambdaFailureFeedbackRoleArn': 'arn:aws:iam::504255336111:role/SNS_CloudWatch_Logs_Access_Role'
}
# Create topic using wrapper
topic = sns_wrapper.create_topic(name='test-sns-functionality', attributes=attributes)
print(topic)
# Lambda ARN (to be subscriber)
lambda_arn = 'arn:aws:lambda:us-east-1:504255336111:function:sns-test-lambda'
# Subscribe lambda to created topic
subscriber_arn = sns_wrapper.subscribe(topic, 'lambda', lambda_arn)
# Add permissions for SNS to invoke lambda
lambdas.add_permission(
FunctionName='sns-test-lambda',
StatementId='permission-for-sns-to-invoke-lambda',
Principal='sns.amazonaws.com',
SourceArn=topic.arn,
Action='lambda:InvokeFunction'
)
# Publish message to topic
sns_wrapper.publish_message(topic, f'This is message from topic: {topic}', attributes={'atr1': 'val1'})
Hope someone will find it helpful!
Upvotes: 0
Related Questions
- AWS SNS ought to trigger my lambda, but does not
- AWS SNS not attaching a Lambda trigger when the subscription is created via the SDK
- Cannot setup aws SNS trigger Lambda function
- Why do I get an error when trying to add an SNS trigger to my AWS Lambda function?
- AWS lambda subscribed to SNS topic not working
- SNS trigger issue for Lambda
- AWS: Publish SNS message for Lambda function via boto3 (Python2)
- Can't add s3 notification for lambda using boto3
- Boto3 SNS inconsistent behavior publishing text message
- Subscribe an Amazon SNS topic using AWS Lambda function?