sns_webhook.py

# 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"

PREFIX = environ.get("ENV_PREFIX", "")
URL = environ.get(PREFIX + "URL")
TEMPLATE = environ.get(PREFIX + "TEMPLATE", "{}")
MESSAGE_KEY = environ.get(PREFIX + "MESSAGE_KEY", "text")
TOPIC_KEY = environ.get(PREFIX + "TOPIC_KEY")


def lambda_handler(event, context):
    """Lambda handler - expects an SNS event"""
    user_agent = context.function_name

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

    # Pretty-print if the message is JSON
    try:
        message = json.dumps(json.loads(message), indent=4)
    except json.decoder.JSONDecodeError:
        pass

    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:
        if response.getheader("content-type").split(";")[0] == "application/json":
            # Round-trip the response to clean it up for the log
            print(json.dumps(json.load(response)))
        else:
            print(response.read().decode("utf-8"))