Andrew Jorgensen
It's better than bad, it's good!

SNS to Webhook Lambda Function

I got an email from AWS telling me, in essence, that I'd been assigned a toll-free number as a unique Origination ID for the SMS messages some of my Lambda functions send me through Amazon SNS, and that in 12 months they'd start billing me $2/month for that number. Since I'm currently paying only $0.37/month for my total AWS usage, that would be a 911% increase!

Fortunately, you can find your Origination ID in the Amazon Pinpoint console and delete it. You probably want to delete your SMS subscriptions first, because you may be assigned a new one as soon as SNS tries to send you another message. That should solve the "I don't want to pay for this" problem, but it doesn't solve the "so, now how do I get notified" problem.

One option is to use a Webhook for some other service you already get notifications from, like Amazon Chime, Google Chat, Microsoft Teams, or Slack (or Matrix, if you don't mind using a third-party bridge). Webhooks are simple API endpoints that use a token and other identifiers in the request URL, and accept some data (usually JSON) to do something - in this case to post a message to a channel or room.

SNS can't call a webhook directly, but you can write a Lambda function that will take an SNS event as input and call your Webhook with your SNS message content. Below is my own little attempt.

# Copyright Andrew Jorgensen
# SPDX-License-Identifier: MIT
"""Receive SNS events in Lambda and POST to a JSON Webhook.

Environment variables required:
* URL - The Webhook URL to POST to (including any required keys)
* TEMPLATE (default: {}) - The JSON data template to POST to the Webhook
* MESSAGE_KEY (default: text) - Key to set to the SNS Message
* TOPIC_KEY (optional) - Key to set to the Topic name from the SNS event
"""
import json
from os import environ
from urllib.request import urlopen, Request

CONTENT_TYPE = "application/json; charset=utf-8"


def lambda_handler(event, context):
    """Lambda handler - expects an SNS event"""
    user_agent = context.function_name
    print(json.dumps(dict(environ), sort_keys=True))
    url = environ.get("URL")
    template = environ.get("TEMPLATE", "{}")
    message_key = environ.get("MESSAGE_KEY", "text")
    topic_key = environ.get("TOPIC_KEY")

    print(json.dumps(event, sort_keys=True))
    topic = event["Records"][0]["Sns"]["TopicArn"].rsplit(":", 1)[1]
    subject = event["Records"][0]["Sns"]["Subject"]
    message = event["Records"][0]["Sns"]["Message"]

    data = json.loads(template)
    if topic_key:
        data[topic_key] = topic
    if subject:
        data[message_key] = f"{subject}: {message}"
    else:
        data[message_key] = message
    data = json.dumps(data, sort_keys=True)
    print(data)

    request = Request(
        url=url,
        data=data.encode("utf-8"),
        headers={"User-Agent": user_agent, "Content-Type": CONTENT_TYPE},
    )
    with urlopen(request) as response:
        print(response.read().decode("utf-8"))

Find current versions on GitHub

At a minimum it needs a URL environment variable with the Inbound Webhook URL you generated. All of Chat, Teams, Slack and Matrix (via t2bot.io) work with just the URL variable set. Amazon Chime requires that you also set MESSAGE_KEY to Content.