Cómo crear un endpoint para recibir notificaciones de SNS con Python

En el post anterior vimos cómo conectar SNS con SES para recibir notificaciones de Bounce, Complaint y Delivery. Aquí construimos el endpoint HTTP que recibe esas notificaciones.

SNS envía los mensajes como peticiones POST a una URL que nosotros definimos. El endpoint necesita:

Confirmar la suscripción cuando SNS lo solicite
Procesar los mensajes de notificación (Bounce, Complaint, Delivery)

Setup

pip install flask requests

bash

Estructura de un mensaje SNS

SNS envía un JSON con el campo Type que indica el tipo de mensaje:

{
  "Type": "SubscriptionConfirmation",
  "MessageId": "...",
  "Token": "...",
  "TopicArn": "arn:aws:sns:us-east-1:...",
  "Message": "You have chosen to subscribe...",
  "SubscribeURL": "https://sns.us-east-1.amazonaws.com/?Action=ConfirmSubscription&...",
  "Timestamp": "2020-04-10T12:00:00.000Z",
  "SignatureVersion": "1",
  "Signature": "...",
  "SigningCertURL": "..."
}

json

Para notificaciones (Type: "Notification"), el campo Message contiene otro JSON con el evento de SES:

{
  "notificationType": "Bounce",
  "bounce": {
    "bounceType": "Permanent",
    "bounceSubType": "General",
    "bouncedRecipients": [
      {
        "emailAddress": "usuario@ejemplo.com",
        "action": "failed",
        "status": "5.1.1",
        "diagnosticCode": "smtp; 550 5.1.1 user unknown"
      }
    ],
    "timestamp": "2020-04-10T12:00:00.000Z"
  },
  "mail": {
    "timestamp": "2020-04-10T11:59:59.000Z",
    "messageId": "...",
    "source": "remitente@tudominio.com"
  }
}

json

Endpoint con Flask

import json
import requests
from flask import Flask, request, jsonify
 
app = Flask(__name__)
 
 
@app.route("/sns/notifications", methods=["POST"])
def sns_endpoint():
    body = request.get_json(force=True)
    message_type = request.headers.get("X-Amz-Sns-Message-Type", "")
 
    if message_type == "SubscriptionConfirmation":
        _confirm_subscription(body)
        return jsonify({"status": "subscribed"}), 200
 
    if message_type == "Notification":
        _handle_notification(body)
        return jsonify({"status": "ok"}), 200
 
    return jsonify({"status": "ignored"}), 200
 
 
def _confirm_subscription(body: dict) -> None:
    """SNS envía una URL para confirmar la suscripción — hay que hacer GET a esa URL."""
    subscribe_url = body.get("SubscribeURL")
    if subscribe_url:
        response = requests.get(subscribe_url, timeout=10)
        print(f"Suscripción confirmada: {response.status_code}")
 
 
def _handle_notification(body: dict) -> None:
    message = json.loads(body.get("Message", "{}"))
    notification_type = message.get("notificationType")
 
    handlers = {
        "Bounce": _handle_bounce,
        "Complaint": _handle_complaint,
        "Delivery": _handle_delivery,
    }
 
    handler = handlers.get(notification_type)
    if handler:
        handler(message)
    else:
        print(f"Tipo de notificación desconocido: {notification_type}")
 
 
def _handle_bounce(message: dict) -> None:
    bounce = message.get("bounce", {})
    bounce_type = bounce.get("bounceType")  # Permanent | Transient
 
    for recipient in bounce.get("bouncedRecipients", []):
        email = recipient["emailAddress"]
        status = recipient.get("status", "")
        print(f"Bounce ({bounce_type}): {email} — {status}")
 
        if bounce_type == "Permanent":
            # Marcar el correo como inválido en la base de datos
            # para no volver a intentar enviarle
            _mark_email_as_bounced(email)
 
 
def _handle_complaint(message: dict) -> None:
    complaint = message.get("complaint", {})
 
    for recipient in complaint.get("complainedRecipients", []):
        email = recipient["emailAddress"]
        print(f"Complaint (spam): {email}")
        # Dar de baja al usuario de las listas de envío
        _unsubscribe_email(email)
 
 
def _handle_delivery(message: dict) -> None:
    delivery = message.get("delivery", {})
    timestamp = delivery.get("timestamp")
 
    for recipient in delivery.get("recipients", []):
        print(f"Delivery OK: {recipient} — {timestamp}")
 
 
def _mark_email_as_bounced(email: str) -> None:
    # Implementar: actualizar BD, deshabilitar correo, etc.
    print(f"[DB] Marcando como bounce: {email}")
 
 
def _unsubscribe_email(email: str) -> None:
    # Implementar: remover de listas de marketing
    print(f"[DB] Dando de baja: {email}")
 
 
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)

python

Suscribir el endpoint en SNS

Con el servidor corriendo y accesible públicamente (ngrok sirve para desarrollo):

# Exponer el servidor local con ngrok para pruebas
ngrok http 5000

bash

Luego en la consola de SNS, al crear la suscripción usar la URL pública:

https://abc123.ngrok.io/sns/notifications

plaintext

SNS enviará inmediatamente un SubscriptionConfirmation — el endpoint debe responder confirmando.

pip install aws-message-validator

bash

from aws_message_validator import MessageValidator
 
validator = MessageValidator()
 
@app.route("/sns/notifications", methods=["POST"])
def sns_endpoint():
    body = request.get_json(force=True)
    try:
        validator.validate(body)  # Lanza excepción si la firma no es válida
    except Exception:
        return jsonify({"error": "invalid signature"}), 403
    # ... resto del handler

python

Setup

Estructura de un mensaje SNS

Endpoint con Flask

Suscribir el endpoint en SNS

Referencias