""" Pruebas del webhook de estado de Twilio. El endpoint es publico, asi que su unica defensa es la firma. Sin ella, cualquiera podria marcar como entregados mensajes que rebotaron, o al reves, y corromper la auditoria. """ from __future__ import annotations import pytest from django.urls import reverse from twilio.request_validator import RequestValidator from notifications.models.event import NotificationEvent from notifications.models.message import Channel, MessageStatus, NotificationMessage pytestmark = pytest.mark.django_db TOKEN = "token-de-pruebas-de-twilio" @pytest.fixture def envio(db): return NotificationMessage.objects.create( channel=Channel.SMS, provider="twilio", status=MessageStatus.SENT, to=["+525512345678"], recipient_count=1, provider_message_id="SM1234567890abcdef", ) @pytest.fixture(autouse=True) def token_configurado(settings): settings.TWILIO_AUTH_TOKEN = TOKEN settings.TWILIO_STATUS_CALLBACK_BASE_URL = "https://notificaciones.ejemplo.com" def publicar(api, envio, parametros, *, firma=None, url=None): ruta = reverse("v1:twilio-webhook", args=[envio.id]) completa = url or f"https://notificaciones.ejemplo.com{ruta}" if firma is None: firma = RequestValidator(TOKEN).compute_signature(completa, parametros) return api.post( ruta, parametros, format="multipart", HTTP_X_TWILIO_SIGNATURE=firma, ) def parametros(**extra) -> dict: return { "MessageSid": "SM1234567890abcdef", "MessageStatus": "delivered", "To": "+525512345678", **extra, } # --------------------------------------------------------------------------- # Firma # --------------------------------------------------------------------------- def test_una_firma_valida_se_acepta(api, envio): respuesta = publicar(api, envio, parametros()) assert respuesta.status_code == 204 assert NotificationEvent.objects.count() == 1 def test_una_firma_invalida_se_rechaza(api, envio): respuesta = publicar(api, envio, parametros(), firma="firmaFalsa=") assert respuesta.status_code == 401 assert NotificationEvent.objects.count() == 0 def test_sin_cabecera_de_firma_se_rechaza(api, envio): ruta = reverse("v1:twilio-webhook", args=[envio.id]) respuesta = api.post(ruta, parametros(), format="multipart") assert respuesta.status_code == 401 def test_una_firma_de_otro_token_se_rechaza(api, envio): ruta = reverse("v1:twilio-webhook", args=[envio.id]) otra = RequestValidator("otro-token").compute_signature( f"https://notificaciones.ejemplo.com{ruta}", parametros() ) assert publicar(api, envio, parametros(), firma=otra).status_code == 401 def test_alterar_un_parametro_invalida_la_firma(api, envio): """La firma cubre los parametros, no solo la URL.""" ruta = reverse("v1:twilio-webhook", args=[envio.id]) firma = RequestValidator(TOKEN).compute_signature( f"https://notificaciones.ejemplo.com{ruta}", parametros() ) respuesta = publicar( api, envio, parametros(MessageStatus="failed"), firma=firma ) assert respuesta.status_code == 401 def test_sin_token_configurado_se_rechaza_todo(api, envio, settings): """ Aceptar avisos "porque aun no esta configurado el token" dejaria la auditoria abierta a cualquiera en internet. """ settings.TWILIO_AUTH_TOKEN = "" assert publicar(api, envio, parametros()).status_code == 401 # --------------------------------------------------------------------------- # Procesamiento # --------------------------------------------------------------------------- def test_un_aviso_de_entrega_actualiza_el_estado(api, envio): publicar(api, envio, parametros(MessageStatus="delivered")) envio.refresh_from_db() assert envio.status == MessageStatus.DELIVERED assert envio.last_event_at is not None def test_un_fallo_registra_el_codigo_de_error(api, envio): publicar( api, envio, parametros( MessageStatus="undelivered", ErrorCode="30006", ErrorMessage="Landline or unreachable carrier", ), ) envio.refresh_from_db() assert envio.status == MessageStatus.BOUNCED assert envio.error_code == "30006" assert "Landline" in envio.error_message def test_el_mismo_aviso_dos_veces_solo_se_guarda_una(api, envio): """Twilio reintenta el callback si no respondemos rapido.""" publicar(api, envio, parametros()) publicar(api, envio, parametros()) assert NotificationEvent.objects.count() == 1 def test_un_aviso_no_retrocede_el_estado(api, envio): """ Los avisos pueden llegar desordenados. Un 'queued' tardio no debe hacer retroceder un envio que ya consta como enviado. """ publicar(api, envio, parametros(MessageStatus="queued")) envio.refresh_from_db() assert envio.status == MessageStatus.SENT def test_un_estado_final_no_se_degrada(api, envio): publicar(api, envio, parametros(MessageStatus="delivered")) publicar(api, envio, parametros(MessageStatus="sent")) envio.refresh_from_db() assert envio.status == MessageStatus.DELIVERED def test_un_aviso_para_un_envio_inexistente_no_revienta(api, envio): import uuid inexistente = uuid.uuid4() ruta = reverse("v1:twilio-webhook", args=[inexistente]) firma = RequestValidator(TOKEN).compute_signature( f"https://notificaciones.ejemplo.com{ruta}", parametros() ) respuesta = api.post( ruta, parametros(), format="multipart", HTTP_X_TWILIO_SIGNATURE=firma ) assert respuesta.status_code == 204 assert NotificationEvent.objects.count() == 0 # --------------------------------------------------------------------------- # Lo firmado tiene que ser lo procesado # --------------------------------------------------------------------------- def _multipart_con_fichero(campos: dict, nombre: str, valor: str) -> tuple[bytes, str]: """ Cuerpo multipart con los campos normales mas una parte con `filename`. Django manda a `request.FILES` toda parte que lleve `filename`, y `FILES` queda fuera de `request.POST`, que es lo unico que firma Twilio. """ limite = "----------------------------pruebadelimite" partes = [] for clave, dato in campos.items(): partes.append( f"--{limite}\r\n" f'Content-Disposition: form-data; name="{clave}"\r\n\r\n{dato}\r\n' ) partes.append( f"--{limite}\r\n" f'Content-Disposition: form-data; name="{nombre}"; filename="{valor}"\r\n' f"Content-Type: text/plain\r\n\r\n{valor}\r\n" ) partes.append(f"--{limite}--\r\n") return "".join(partes).encode(), f"multipart/form-data; boundary={limite}" def test_una_parte_con_fichero_no_puede_cambiar_lo_firmado(api, envio): """ Con una firma capturada se podia reenviar la peticion anadiendo una parte `name="MessageStatus"; filename="failed"`. La firma seguia cuadrando —se calcula sobre `request.POST`, donde esa parte no aparece— pero la vista leia `request.data`, que fusiona `request.FILES` encima y se queda con el ultimo valor. El estado que la firma protegia acababa cambiado. """ ruta = reverse("v1:twilio-webhook", args=[envio.id]) completa = f"https://notificaciones.ejemplo.com{ruta}" honestos = parametros() # MessageStatus = "delivered" firma = RequestValidator(TOKEN).compute_signature(completa, honestos) cuerpo, tipo = _multipart_con_fichero(honestos, "MessageStatus", "failed") respuesta = api.post( ruta, cuerpo, content_type=tipo, HTTP_X_TWILIO_SIGNATURE=firma ) # Se rechaza: un callback de estado nunca lleva ficheros. assert respuesta.status_code == 401 assert NotificationEvent.objects.count() == 0 envio.refresh_from_db() assert envio.status == MessageStatus.SENT def test_un_campo_no_firmado_no_entra(api, envio): """Una clave que no estaba en lo firmado tampoco debe colarse.""" ruta = reverse("v1:twilio-webhook", args=[envio.id]) completa = f"https://notificaciones.ejemplo.com{ruta}" honestos = parametros() firma = RequestValidator(TOKEN).compute_signature(completa, honestos) cuerpo, tipo = _multipart_con_fichero(honestos, "ErrorCode", "30008") respuesta = api.post( ruta, cuerpo, content_type=tipo, HTTP_X_TWILIO_SIGNATURE=firma ) assert respuesta.status_code == 401