django-anymail/anymail/webhooks/resend.py

import json
from datetime import datetime

from ..exceptions import (
    AnymailImproperlyInstalled,
    AnymailInvalidAddress,
    AnymailWebhookValidationFailure,
    _LazyError,
)
from ..signals import AnymailTrackingEvent, EventType, RejectReason, tracking
from ..utils import get_anymail_setting, parse_single_address
from .base import AnymailBaseWebhookView, AnymailCoreWebhookView

try:
    # Valid webhook signatures with svix library if available
    from svix.webhooks import Webhook as SvixWebhook, WebhookVerificationError
except ImportError:
    # Otherwise, validating with basic auth is sufficient
    # (unless settings specify signature validation, which will then raise this error)
    SvixWebhook = _LazyError(
        AnymailImproperlyInstalled(missing_package="svix", install_extra="resend")
    )
    WebhookVerificationError = object()


class SvixWebhookValidationMixin(AnymailCoreWebhookView):
    """Mixin to validate Svix webhook signatures"""

    # Consuming classes can override (e.g., to use different secrets
    # for inbound and tracking webhooks).
    _secret_setting_name = "signing_secret"

    @classmethod
    def as_view(cls, **initkwargs):
        if not hasattr(cls, cls._secret_setting_name):
            # The attribute must exist on the class before View.as_view
            # will allow overrides via kwarg
            setattr(cls, cls._secret_setting_name, None)
        return super().as_view(**initkwargs)

    def __init__(self, **kwargs):
        self.signing_secret = get_anymail_setting(
            self._secret_setting_name,
            esp_name=self.esp_name,
            default=None,
            kwargs=kwargs,
        )
        if self.signing_secret is None:
            self._svix_webhook = None
            self.warn_if_no_basic_auth = True
        else:
            # This will raise an import error if svix isn't installed
            self._svix_webhook = SvixWebhook(self.signing_secret)
            # Basic auth is not required if validating signature
            self.warn_if_no_basic_auth = False
        super().__init__(**kwargs)

    def validate_request(self, request):
        if self._svix_webhook:
            # https://docs.svix.com/receiving/verifying-payloads/how
            try:
                # Note: if signature is valid, Svix also tries to parse
                # the json body, so this could raise other errors...
                self._svix_webhook.verify(request.body, request.headers)
            except WebhookVerificationError as error:
                setting_name = f"{self.esp_name}_{self._secret_setting_name}".upper()
                raise AnymailWebhookValidationFailure(
                    f"{self.esp_name} webhook called with incorrect signature"
                    f" (check Anymail {setting_name} setting)"
                ) from error


class ResendTrackingWebhookView(SvixWebhookValidationMixin, AnymailBaseWebhookView):
    """Handler for Resend.com status tracking webhooks"""

    esp_name = "Resend"
    signal = tracking

    def parse_events(self, request):
        esp_event = json.loads(request.body.decode("utf-8"))
        return [self.esp_to_anymail_event(esp_event, request)]

    # https://resend.com/docs/dashboard/webhooks/event-types
    event_types = {
        # Map Resend type: Anymail normalized type
        "email.sent": EventType.SENT,
        "email.delivered": EventType.DELIVERED,
        "email.delivery_delayed": EventType.DEFERRED,
        "email.complained": EventType.COMPLAINED,
        "email.bounced": EventType.BOUNCED,
        "email.opened": EventType.OPENED,
        "email.clicked": EventType.CLICKED,
    }

    def esp_to_anymail_event(self, esp_event, request):
        event_type = self.event_types.get(esp_event["type"], EventType.UNKNOWN)

        # event_id: HTTP header `svix-id` is unique for a particular event
        # (including across reposts due to errors)
        try:
            event_id = request.headers["svix-id"]
        except KeyError:
            event_id = None

        # timestamp: Payload created_at is unique for a particular event.
        # (Payload data.created_at is when the message was created, not the event.
        # HTTP header `svix-timestamp` changes for each repost of the same event.)
        try:
            timestamp = datetime.fromisoformat(
                # Must convert "Z" to timezone offset for Python 3.10 and earlier.
                esp_event["created_at"].replace("Z", "+00:00")
            )
        except (KeyError, ValueError):
            timestamp = None

        try:
            message_id = esp_event["data"]["email_id"]
        except (KeyError, TypeError):
            message_id = None

        # Resend doesn't provide bounce reasons or SMTP responses,
        # but it's possible to distinguish some cases by examining
        # the human-readable message text:
        try:
            bounce_message = esp_event["data"]["bounce"]["message"]
        except (KeyError, ValueError):
            bounce_message = None
            reject_reason = None
        else:
            if "suppressed sending" in bounce_message:
                # "Resend has suppressed sending to this address ..."
                reject_reason = RejectReason.BLOCKED
            elif "bounce message" in bounce_message:
                # "The recipient's email provider sent a hard bounce message, ..."
                # "The recipient's email provider sent a general bounce message. ..."
                # "The recipient's email provider sent a bounce message because
                #    the recipient's inbox was full. ..."
                reject_reason = RejectReason.BOUNCED
            else:
                reject_reason = RejectReason.OTHER  # unknown

        # Recover tags and metadata from custom headers
        metadata = {}
        tags = []
        try:
            headers = esp_event["data"]["headers"]
        except KeyError:
            pass
        else:
            for header in headers:
                name = header["name"].lower()
                if name == "x-tags":
                    try:
                        tags = json.loads(header["value"])
                    except (ValueError, TypeError):
                        pass
                elif name == "x-metadata":
                    try:
                        metadata = json.loads(header["value"])
                    except (ValueError, TypeError):
                        pass

        # For multi-recipient emails (including cc and bcc), Resend generates events
        # for each recipient, but no indication of which recipient an event applies to.
        # Just report the first `to` recipient.
        try:
            first_to = esp_event["data"]["to"][0]
            recipient = parse_single_address(first_to).addr_spec
        except (KeyError, IndexError, TypeError, AnymailInvalidAddress):
            recipient = None

        try:
            click_data = esp_event["data"]["click"]
        except (KeyError, TypeError):
            click_url = None
            user_agent = None
        else:
            click_url = click_data.get("link")
            user_agent = click_data.get("userAgent")

        return AnymailTrackingEvent(
            event_type=event_type,
            timestamp=timestamp,
            message_id=message_id,
            event_id=event_id,
            recipient=recipient,
            reject_reason=reject_reason,
            description=bounce_message,
            mta_response=None,
            tags=tags,
            metadata=metadata,
            click_url=click_url,
            user_agent=user_agent,
            esp_event=esp_event,
        )