""" Pruebas del webhook de eventos de SendGrid. El endpoint es publico, asi que su unica defensa es la firma. """ from __future__ import annotations import json import time import pytest from notifications.channels.email.sendgrid import CUSTOM_ARG_MESSAGE_ID from notifications.models.event import NotificationEvent from notifications.models.message import ( Channel, MessageStatus, NotificationMessage, ) from notifications.tests.conftest import firmar_webhook pytestmark = pytest.mark.django_db URL = "/notifications/v1/webhooks/sendgrid" @pytest.fixture def envio(db): return NotificationMessage.objects.create( channel=Channel.EMAIL, provider="sendgrid", status=MessageStatus.SENT, subject="Prueba", sender="avisos@ejemplo.com", to=["destino@ejemplo.com"], recipient_count=1, provider_message_id="sgmsg123", ) def evento(envio=None, **extra) -> dict: base = { "email": "destino@ejemplo.com", "timestamp": int(time.time()), "event": "delivered", "sg_event_id": "evt-0001", "sg_message_id": "sgmsg123.filterdrecv-abc", } if envio is not None: base[CUSTOM_ARG_MESSAGE_ID] = str(envio.id) return {**base, **extra} def publicar(api, eventos, privada, *, timestamp=None, firma=None): cuerpo = json.dumps(eventos).encode("utf-8") marca = timestamp or str(int(time.time())) return api.post( URL, data=cuerpo, content_type="application/json", HTTP_X_TWILIO_EMAIL_EVENT_WEBHOOK_SIGNATURE=( firma if firma is not None else firmar_webhook(privada, marca, cuerpo) ), HTTP_X_TWILIO_EMAIL_EVENT_WEBHOOK_TIMESTAMP=marca, ) # --------------------------------------------------------------------------- # Firma # --------------------------------------------------------------------------- def test_una_firma_valida_se_acepta(api, settings, claves_webhook, envio): privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica assert publicar(api, [evento(envio)], privada).status_code == 204 assert NotificationEvent.objects.count() == 1 def test_una_firma_invalida_se_rechaza(api, settings, claves_webhook, envio): privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica respuesta = publicar(api, [evento(envio)], privada, firma="ZmlybWFGYWxzYQ==") assert respuesta.status_code == 401 assert NotificationEvent.objects.count() == 0 def test_una_firma_de_otra_clave_se_rechaza(api, settings, claves_webhook, envio): from cryptography.hazmat.primitives.asymmetric import ec _, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica otra = ec.generate_private_key(ec.SECP256R1()) assert publicar(api, [evento(envio)], otra).status_code == 401 def test_sin_cabeceras_de_firma_se_rechaza(api, settings, claves_webhook): _, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica assert api.post(URL, data=b"[]", content_type="application/json").status_code == 401 def test_sin_clave_configurada_se_rechaza_todo(api, settings, claves_webhook, envio): """ Aceptar eventos "porque aun no esta configurada la clave" dejaria la auditoria abierta a cualquiera en internet. """ privada, _ = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = "" assert publicar(api, [evento(envio)], privada).status_code == 401 def test_una_peticion_antigua_se_rechaza(api, settings, claves_webhook, envio): privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica settings.SENDGRID_WEBHOOK_TOLERANCE_SECONDS = 60 viejo = str(int(time.time()) - 3600) assert publicar(api, [evento(envio)], privada, timestamp=viejo).status_code == 401 def test_cambiar_el_cuerpo_invalida_la_firma(api, settings, claves_webhook, envio): privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica marca = str(int(time.time())) original = json.dumps([evento(envio)]).encode("utf-8") firma = firmar_webhook(privada, marca, original) alterado = json.dumps([evento(envio, event="bounce")]).encode("utf-8") respuesta = api.post( URL, data=alterado, content_type="application/json", HTTP_X_TWILIO_EMAIL_EVENT_WEBHOOK_SIGNATURE=firma, HTTP_X_TWILIO_EMAIL_EVENT_WEBHOOK_TIMESTAMP=marca, ) assert respuesta.status_code == 401 # --------------------------------------------------------------------------- # Procesamiento # --------------------------------------------------------------------------- def test_un_evento_de_entrega_actualiza_el_estado(api, settings, claves_webhook, envio): privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica publicar(api, [evento(envio, event="delivered")], privada) envio.refresh_from_db() assert envio.status == MessageStatus.DELIVERED assert envio.last_event_at is not None def test_un_rebote_marca_el_envio_como_rebotado(api, settings, claves_webhook, envio): privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica publicar( api, [evento(envio, event="bounce", sg_event_id="evt-b", reason="550 no existe")], privada, ) envio.refresh_from_db() assert envio.status == MessageStatus.BOUNCED assert envio.events.first().reason == "550 no existe" def test_el_mismo_evento_dos_veces_solo_se_guarda_una( api, settings, claves_webhook, envio ): """SendGrid reenvia el lote si tardamos en responder o damos un 5xx.""" privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica publicar(api, [evento(envio)], privada) publicar(api, [evento(envio)], privada) assert NotificationEvent.objects.count() == 1 def test_un_evento_tardio_no_degrada_un_estado_final( api, settings, claves_webhook, envio ): """Los webhooks no garantizan orden de entrega.""" privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica publicar(api, [evento(envio, event="bounce", sg_event_id="evt-1")], privada) publicar(api, [evento(envio, event="deferred", sg_event_id="evt-2")], privada) envio.refresh_from_db() assert envio.status == MessageStatus.BOUNCED def test_una_apertura_no_cambia_el_estado(api, settings, claves_webhook, envio): privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica publicar(api, [evento(envio, event="open", sg_event_id="evt-open")], privada) envio.refresh_from_db() assert envio.status == MessageStatus.SENT assert envio.events.count() == 1 def test_se_correlaciona_por_message_id_si_falta_el_argumento( api, settings, claves_webhook, envio ): privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica publicar(api, [evento()], privada) # sin CUSTOM_ARG_MESSAGE_ID assert NotificationEvent.objects.get().message_id == envio.id def test_un_evento_sin_identificador_no_se_guarda(api, settings, claves_webhook, envio): """Sin `sg_event_id` no se puede garantizar idempotencia.""" privada, publica = claves_webhook settings.SENDGRID_WEBHOOK_PUBLIC_KEY = publica sin_id = evento(envio) sin_id.pop("sg_event_id") publicar(api, [sin_id], privada) assert NotificationEvent.objects.count() == 0