From 6026821f20a9f0723829b87d8b744e710b6e450a Mon Sep 17 00:00:00 2001 From: Sylvain Zimmer Date: Mon, 30 Mar 2026 12:30:49 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8(inbound)=20add=20prototype=20Brevo=20?= =?UTF-8?q?inbound=20hook=20for=20email=20delivery?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/api/viewsets/inbound/brevo.py | 297 +++++++++++ .../core/tests/api/test_inbound_brevo.py | 487 ++++++++++++++++++ src/backend/core/urls.py | 4 + 3 files changed, 788 insertions(+) create mode 100644 src/backend/core/api/viewsets/inbound/brevo.py create mode 100644 src/backend/core/tests/api/test_inbound_brevo.py diff --git a/src/backend/core/api/viewsets/inbound/brevo.py b/src/backend/core/api/viewsets/inbound/brevo.py new file mode 100644 index 000000000..e700f1adc --- /dev/null +++ b/src/backend/core/api/viewsets/inbound/brevo.py @@ -0,0 +1,297 @@ +"""Brevo inbound channel implementation for handling email from Brevo parse webhooks.""" + +import hashlib +import logging +import secrets +from email.utils import parsedate_to_datetime +from typing import Any, Dict, List, Optional + +from django.conf import settings +from django.core.exceptions import ValidationError +from django.core.validators import validate_email +from django.utils import timezone + +from drf_spectacular.utils import extend_schema +from rest_framework import status, viewsets +from rest_framework.authentication import BaseAuthentication +from rest_framework.decorators import action +from rest_framework.exceptions import AuthenticationFailed +from rest_framework.permissions import IsAuthenticated +from rest_framework.response import Response + +from core import models +from core.mda.inbound import check_local_recipients, deliver_inbound_message +from core.mda.rfc5322 import compose_email + +logger = logging.getLogger(__name__) + + +class BrevoAuthentication(BaseAuthentication): + """ + Custom authentication for Brevo webhook endpoints. + + Supports two authentication methods: + 1. Channel ID authentication (X-Channel-ID header) - for channel-specific webhooks + 2. HMAC signature validation (X-Brevo-Signature header) - for global webhooks + + Returns None or (user, auth) + """ + + def authenticate(self, request): + channel_id = request.headers.get("X-Channel-ID") + if channel_id: + return self._authenticate_by_channel_id(channel_id) + + signature = request.headers.get("X-Brevo-Signature") + if signature: + return self._authenticate_by_signature(request, signature) + + raise AuthenticationFailed("Missing authentication credentials") + + def _authenticate_by_channel_id(self, channel_id: str): + try: + channel = models.Channel.objects.get(id=channel_id, type="brevo") + except (models.Channel.DoesNotExist, ValidationError) as e: + raise AuthenticationFailed("Invalid channel_id") from e + return (None, {"channel": channel, "auth_type": "channel_id"}) + + def _authenticate_by_signature(self, request, signature: str): + secret = getattr(settings, "BREVO_WEBHOOK_SECRET", None) + if not secret: + raise AuthenticationFailed("Brevo webhook secret not configured") + + body = request.body + expected_signature = hashlib.sha256(secret.encode() + body).hexdigest() + if not secrets.compare_digest(signature, expected_signature): + raise AuthenticationFailed("Invalid signature") + + return (None, {"auth_type": "hmac"}) + + def authenticate_header(self, request): + return 'Bearer realm="Brevo"' + + +def convert_brevo_payload_to_parsed_email(item: Dict[str, Any]) -> Dict[str, Any]: + """Convert Brevo parsed email payload to internal email format.""" + from_email = item.get("From", {}) + to_emails = item.get("To", []) + cc_emails = item.get("Cc", []) + reply_to = item.get("ReplyTo") + + subject = item.get("Subject", "(no subject)") + extracted_message = item.get("ExtractedMarkdownMessage", "") + raw_html_body = item.get("RawHtmlBody") + raw_text_body = item.get("RawTextBody") + in_reply_to = item.get("InReplyTo") + sent_at = item.get("SentAtDate") + + html_body = "" + if raw_html_body: + html_body = raw_html_body + elif extracted_message: + html_body = extracted_message.replace("\n", "
") + + text_body = raw_text_body or extracted_message or "" + + parsed_email: Dict[str, Any] = { + "subject": subject, + "from": { + "email": from_email.get("Address", ""), + "name": from_email.get("Name"), + }, + "to": [ + { + "email": to.get("Address", ""), + "name": to.get("Name"), + } + for to in to_emails + ], + "cc": [ + { + "email": cc.get("Address", ""), + "name": cc.get("Name"), + } + for cc in cc_emails + ], + "headers": {}, + } + + if reply_to: + parsed_email["reply_to"] = { + "email": reply_to.get("Address", ""), + "name": reply_to.get("Name"), + } + + if in_reply_to: + parsed_email["in_reply_to"] = in_reply_to + parsed_email["references"] = in_reply_to + + if sent_at: + try: + parsed_email["date"] = parsedate_to_datetime(sent_at) + except (ValueError, TypeError): + parsed_email["date"] = timezone.now() + else: + parsed_email["date"] = timezone.now() + + if html_body: + parsed_email["htmlBody"] = [{"content": html_body}] + if text_body: + parsed_email["textBody"] = [{"content": text_body}] + + return parsed_email + + +class InboundBrevoViewSet(viewsets.GenericViewSet): + """Handles incoming email messages from Brevo inbound parse webhooks.""" + + CHANNEL_TYPE = "brevo" + CHANNEL_DESCRIPTION = "Brevo inbound email parsing" + + permission_classes = [IsAuthenticated] + authentication_classes = [BrevoAuthentication] + + @extend_schema(exclude=True) + @action( + detail=False, + methods=["post"], + url_path="webhook", + url_name="inbound-brevo-webhook", + ) + def webhook(self, request): + """Handle incoming Brevo webhook with parsed email(s).""" + + data = request.data + items = data.get("items", []) + + if not items: + return Response( + {"detail": "No items in payload"}, + status=status.HTTP_400_BAD_REQUEST, + ) + + if not isinstance(items, list): + return Response( + {"detail": "Items must be a list"}, + status=status.HTTP_400_BAD_REQUEST, + ) + + auth_data = request.auth + channel = auth_data.get("channel") + + results = [] + success_count = 0 + failure_count = 0 + + for item in items: + result = self._process_brevo_item(item, channel) + if result["success"]: + success_count += 1 + else: + failure_count += 1 + results.append(result) + + if failure_count > 0 and success_count == 0: + return Response( + { + "status": "error", + "detail": "Failed to process all messages", + "results": results, + }, + status=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + + if failure_count > 0: + return Response( + { + "status": "partial_success", + "processed": success_count, + "failed": failure_count, + "results": results, + }, + status=status.HTTP_207_MULTI_STATUS, + ) + + logger.info( + "Successfully processed %d Brevo inbound messages", + success_count, + ) + return Response( + { + "status": "ok", + "processed": success_count, + "results": results, + } + ) + + def _process_brevo_item(self, item: Dict[str, Any], channel: Optional[models.Channel]) -> Dict[str, Any]: + """Process a single Brevo email item.""" + try: + from_email = item.get("From", {}).get("Address", "") + if not from_email: + return {"success": False, "error": "Missing From address"} + + recipients = item.get("Recipients", []) + if not recipients: + recipients = [r.get("Address") for r in item.get("To", []) if r.get("Address")] + if not recipients: + return {"success": False, "error": "No recipients found"} + + recipient_emails = [r for r in recipients if r] + if not recipient_emails: + return {"success": False, "error": "No valid recipient addresses"} + + local_recipients = check_local_recipients(recipient_emails) + + parsed_email = convert_brevo_payload_to_parsed_email(item) + prepend_headers = [ + ("X-Brevo-Webhook", "inbound"), + ("Received", "from brevo-inbound"), + ] + + raw_email = compose_email(parsed_email, prepend_headers=prepend_headers) + + delivered_count = 0 + for recipient in recipient_emails: + if recipient in local_recipients: + delivered = deliver_inbound_message( + recipient, + parsed_email, + raw_email, + channel=channel, + skip_inbound_queue=True, + ) + if delivered: + delivered_count += 1 + + return { + "success": delivered_count > 0, + "message_id": item.get("MessageId"), + "delivered": delivered_count, + "recipients": len(recipient_emails), + } + + except Exception as e: + logger.exception("Error processing Brevo item: %s", e) + return {"success": False, "error": str(e)} + + @extend_schema(exclude=True) + @action( + detail=False, + methods=["post"], + url_path="check", + url_name="inbound-brevo-check", + ) + def check(self, request): + """Check which recipients are locally deliverable.""" + data = request.data + addresses = data.get("addresses", []) + if not addresses or not isinstance(addresses, list): + return Response( + {"detail": "Missing addresses"}, + status=status.HTTP_400_BAD_REQUEST, + ) + + local_addresses = check_local_recipients(addresses) + results = {address: address in local_addresses for address in addresses} + return Response(results) diff --git a/src/backend/core/tests/api/test_inbound_brevo.py b/src/backend/core/tests/api/test_inbound_brevo.py new file mode 100644 index 000000000..d6dd9f81a --- /dev/null +++ b/src/backend/core/tests/api/test_inbound_brevo.py @@ -0,0 +1,487 @@ +"""Tests for Brevo inbound API endpoints.""" + +import hashlib +import json +from unittest.mock import patch + +import pytest +from django.test import override_settings +from rest_framework import status +from rest_framework.exceptions import AuthenticationFailed +from rest_framework.test import APIClient + +from core import factories, models +from core.api.viewsets.inbound.brevo import ( + BrevoAuthentication, + convert_brevo_payload_to_parsed_email, +) + + +@pytest.fixture(name="api_client") +def fixture_api_client(): + """Return an API client.""" + return APIClient() + + +@pytest.fixture(name="channel") +def fixture_channel(): + """Create a test channel for Brevo.""" + mailbox = factories.MailboxFactory() + return factories.ChannelFactory( + type="brevo", + mailbox=mailbox, + settings={ + "config": {"enabled": True}, + }, + ) + + +@pytest.fixture(name="channel_with_settings") +def fixture_channel_with_settings(): + """Create a test channel with settings.""" + mailbox = factories.MailboxFactory() + return factories.ChannelFactory( + type="brevo", + mailbox=mailbox, + settings={ + "config": {"enabled": True}, + "tags": [], + }, + ) + + +@pytest.fixture(name="brevo_payload") +def fixture_brevo_payload(): + """Return a sample Brevo webhook payload.""" + return { + "items": [ + { + "Uuid": ["1a825d56-029b-4a41-b8e4-61670463431b"], + "MessageId": "", + "InReplyTo": None, + "From": { + "Name": "Test Sender", + "Address": "sender@example.com", + }, + "To": [ + { + "Name": "Test Recipient", + "Address": "recipient@test.com", + } + ], + "Recipients": ["recipient@test.com"], + "Cc": [], + "ReplyTo": None, + "SentAtDate": "Mon, 15 Jan 2024 10:00:00 +0000", + "Subject": "Test Subject", + "RawHtmlBody": "

Test message body

", + "RawTextBody": "Test message body", + "ExtractedMarkdownMessage": "Test message body", + "ExtractedMarkdownSignature": "-- \nTest Sender", + "SpamScore": 1.0, + "Attachments": [], + "Headers": {}, + } + ] + } + + +class TestConvertBrevoPayload: + """Test the Brevo payload conversion function.""" + + def test_convert_full_payload(self): + """Test conversion of a full Brevo payload.""" + item = { + "From": {"Address": "sender@example.com", "Name": "Sender"}, + "To": [{"Address": "recipient@example.com", "Name": "Recipient"}], + "Cc": [{"Address": "cc@example.com", "Name": "CC"}], + "Subject": "Test Subject", + "ExtractedMarkdownMessage": "Hello World", + "RawHtmlBody": "

Hello World

", + "RawTextBody": "Hello World", + "SentAtDate": "Mon, 15 Jan 2024 10:00:00 +0000", + "InReplyTo": "", + } + + result = convert_brevo_payload_to_parsed_email(item) + + assert result["subject"] == "Test Subject" + assert result["from"]["email"] == "sender@example.com" + assert result["from"]["name"] == "Sender" + assert len(result["to"]) == 1 + assert result["to"][0]["email"] == "recipient@example.com" + assert result["in_reply_to"] == "" + assert "htmlBody" in result + assert "textBody" in result + + def test_convert_minimal_payload(self): + """Test conversion with minimal payload.""" + item = { + "From": {"Address": "sender@example.com"}, + "Subject": "Test", + } + + result = convert_brevo_payload_to_parsed_email(item) + + assert result["subject"] == "Test" + assert result["from"]["email"] == "sender@example.com" + assert result["from"]["name"] is None + + def test_convert_without_html_uses_markdown(self): + """Test that markdown is used when HTML is not available.""" + item = { + "From": {"Address": "sender@example.com"}, + "Subject": "Test", + "ExtractedMarkdownMessage": "Markdown **content**", + } + + result = convert_brevo_payload_to_parsed_email(item) + + assert "htmlBody" in result + assert "**content**" in result["htmlBody"][0]["content"] + + +@pytest.mark.django_db +class TestBrevoAuthentication: + """Test the BrevoAuthentication class.""" + + def test_authenticate_with_valid_channel_id(self, channel): + """Test authentication with valid channel ID.""" + auth = BrevoAuthentication() + + class MockRequest: + """Mock request.""" + + def __init__(self, channel_id): + self.headers = {"X-Channel-ID": str(channel_id)} + + request = MockRequest(channel.id) + user, auth_data = auth.authenticate(request) + + assert user is None + assert auth_data["channel"] == channel + assert auth_data["auth_type"] == "channel_id" + + def test_authenticate_with_missing_channel_id(self): + """Test authentication fails with missing credentials.""" + auth = BrevoAuthentication() + + class MockRequest: + """Mock request.""" + + def __init__(self): + self.headers = {} + + request = MockRequest() + + with pytest.raises(AuthenticationFailed, match="Missing authentication"): + auth.authenticate(request) + + def test_authenticate_with_invalid_channel_id(self): + """Test authentication fails with invalid channel ID.""" + auth = BrevoAuthentication() + + class MockRequest: + """Mock request.""" + + def __init__(self): + self.headers = {"X-Channel-ID": "invalid-uuid"} + + request = MockRequest() + + with pytest.raises(AuthenticationFailed, match="Invalid channel_id"): + auth.authenticate(request) + + @override_settings(BREVO_WEBHOOK_SECRET="test-secret") + def test_authenticate_with_valid_hmac(self): + """Test authentication with valid HMAC signature.""" + auth = BrevoAuthentication() + + secret = "test-secret" + body = b'{"items": []}' + signature = hashlib.sha256(secret.encode() + body).hexdigest() + + class MockRequest: + """Mock request.""" + + def __init__(self): + self.headers = {"X-Brevo-Signature": signature} + self.body = body + + request = MockRequest() + user, auth_data = auth.authenticate(request) + + assert user is None + assert auth_data["auth_type"] == "hmac" + + @override_settings(BREVO_WEBHOOK_SECRET="test-secret") + def test_authenticate_with_invalid_hmac(self): + """Test authentication fails with invalid HMAC signature.""" + auth = BrevoAuthentication() + + class MockRequest: + """Mock request.""" + + def __init__(self): + self.headers = {"X-Brevo-Signature": "invalid-signature"} + self.body = b'{"items": []}' + + request = MockRequest() + + with pytest.raises(AuthenticationFailed, match="Invalid signature"): + auth.authenticate(request) + + def test_authenticate_without_secret(self): + """Test authentication fails when secret is not configured.""" + auth = BrevoAuthentication() + + class MockRequest: + """Mock request.""" + + def __init__(self): + self.headers = {"X-Brevo-Signature": "some-signature"} + self.body = b'{"items": []}' + + request = MockRequest() + + with pytest.raises(AuthenticationFailed, match="not configured"): + auth.authenticate(request) + + +@pytest.mark.django_db +class TestInboundBrevoWebhook: + """Test the webhook endpoint.""" + + @patch("core.api.viewsets.inbound.brevo.deliver_inbound_message") + def test_webhook_success(self, mock_deliver, api_client, channel, brevo_payload): + """Test successful message delivery.""" + mock_deliver.return_value = True + + response = api_client.post( + "/api/v1.0/inbound/brevo/webhook/", + data=brevo_payload, + format="json", + HTTP_X_CHANNEL_ID=str(channel.id), + ) + + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert data["status"] == "ok" + assert data["processed"] == 1 + + def test_webhook_empty_items(self, api_client, channel): + """Test webhook with empty items list.""" + response = api_client.post( + "/api/v1.0/inbound/brevo/webhook/", + data={"items": []}, + format="json", + HTTP_X_CHANNEL_ID=str(channel.id), + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"] == "No items in payload" + + def test_webhook_missing_items(self, api_client, channel): + """Test webhook with missing items field.""" + response = api_client.post( + "/api/v1.0/inbound/brevo/webhook/", + data={}, + format="json", + HTTP_X_CHANNEL_ID=str(channel.id), + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + + def test_webhook_invalid_items_type(self, api_client, channel): + """Test webhook with invalid items type.""" + response = api_client.post( + "/api/v1.0/inbound/brevo/webhook/", + data={"items": "not a list"}, + format="json", + HTTP_X_CHANNEL_ID=str(channel.id), + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json()["detail"] == "Items must be a list" + + def test_webhook_without_authentication(self, api_client, brevo_payload): + """Test webhook endpoint without authentication.""" + response = api_client.post( + "/api/v1.0/inbound/brevo/webhook/", + data=brevo_payload, + format="json", + ) + + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_webhook_item_missing_from(self, api_client, channel): + """Test webhook with item missing From address.""" + payload = { + "items": [ + { + "To": [{"Address": "recipient@test.com"}], + "Subject": "Test", + } + ] + } + + response = api_client.post( + "/api/v1.0/inbound/brevo/webhook/", + data=payload, + format="json", + HTTP_X_CHANNEL_ID=str(channel.id), + ) + + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert data["processed"] == 0 + assert data["results"][0]["success"] is False + + +@pytest.mark.django_db +class TestInboundBrevoCheck: + """Test the check endpoint.""" + + def test_check_success(self, api_client, channel): + """Test successful recipient check.""" + response = api_client.post( + "/api/v1.0/inbound/brevo/check/", + data={"addresses": ["recipient@test.com"]}, + format="json", + HTTP_X_CHANNEL_ID=str(channel.id), + ) + + assert response.status_code == status.HTTP_200_OK + assert response.json()["recipient@test.com"] is True + + def test_check_missing_addresses(self, api_client, channel): + """Test check with missing addresses.""" + response = api_client.post( + "/api/v1.0/inbound/brevo/check/", + data={}, + format="json", + HTTP_X_CHANNEL_ID=str(channel.id), + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + + def test_check_empty_addresses(self, api_client, channel): + """Test check with empty addresses list.""" + response = api_client.post( + "/api/v1.0/inbound/brevo/check/", + data={"addresses": []}, + format="json", + HTTP_X_CHANNEL_ID=str(channel.id), + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + + +@pytest.mark.django_db +class TestInboundBrevoE2E: + """End-to-end tests for Brevo inbound channel.""" + + def test_webhook_message_e2e(self, api_client): + """Test that message is properly created from Brevo payload.""" + assert models.Message.objects.count() == 0 + + mailbox = factories.MailboxFactory() + label = factories.LabelFactory(mailbox=mailbox, name="Brevo") + channel = factories.ChannelFactory( + type="brevo", + mailbox=mailbox, + settings={ + "config": {"enabled": True}, + "tags": [str(label.id)], + }, + ) + + payload = { + "items": [ + { + "MessageId": "", + "From": { + "Name": "External Sender", + "Address": "external@example.com", + }, + "To": [ + { + "Name": mailbox.address, + "Address": mailbox.address, + } + ], + "Recipients": [mailbox.address], + "Cc": [], + "SentAtDate": "Mon, 15 Jan 2024 10:00:00 +0000", + "Subject": "Brevo Test Email", + "RawHtmlBody": "

Hello from Brevo

", + "RawTextBody": "Hello from Brevo", + "ExtractedMarkdownMessage": "Hello from Brevo", + "Attachments": [], + "Headers": {}, + } + ] + } + + response = api_client.post( + "/api/v1.0/inbound/brevo/webhook/", + data=payload, + format="json", + HTTP_X_CHANNEL_ID=str(channel.id), + ) + + assert response.status_code == status.HTTP_200_OK + + assert models.Message.objects.count() == 1 + message = models.Message.objects.first() + + assert message.thread.accesses.first().mailbox == mailbox + assert message.sender.email == "external@example.com" + assert message.subject == "Brevo Test Email" + + thread_label_ids = set(message.thread.labels.values_list("id", flat=True)) + assert label.id in thread_label_ids + + def test_webhook_with_reply(self, api_client): + """Test that reply is properly threaded.""" + mailbox = factories.MailboxFactory() + channel = factories.ChannelFactory( + type="brevo", + mailbox=mailbox, + ) + + existing_message = factories.MessageFactory( + thread__mailbox=mailbox, + mime_id="original-message-id@example.com", + ) + + payload = { + "items": [ + { + "MessageId": "", + "InReplyTo": "", + "From": { + "Address": "reply@example.com", + }, + "To": [{"Address": mailbox.address}], + "Recipients": [mailbox.address], + "Subject": "Re: Original Subject", + "ExtractedMarkdownMessage": "This is a reply", + "SentAtDate": "Mon, 15 Jan 2024 11:00:00 +0000", + } + ] + } + + response = api_client.post( + "/api/v1.0/inbound/brevo/webhook/", + data=payload, + format="json", + HTTP_X_CHANNEL_ID=str(channel.id), + ) + + assert response.status_code == status.HTTP_200_OK + + assert models.Message.objects.count() == 2 + new_message = models.Message.objects.exclude(id=existing_message.id).first() + + assert new_message.thread == existing_message.thread diff --git a/src/backend/core/urls.py b/src/backend/core/urls.py index cb482f4d4..63544eb87 100644 --- a/src/backend/core/urls.py +++ b/src/backend/core/urls.py @@ -16,6 +16,7 @@ from core.api.viewsets.import_message import ImportViewSet, MessagesArchiveUploadViewSet from core.api.viewsets.inbound.mta import InboundMTAViewSet from core.api.viewsets.inbound.widget import InboundWidgetViewSet +from core.api.viewsets.inbound.brevo import InboundBrevoViewSet from core.api.viewsets.label import LabelViewSet from core.api.viewsets.mailbox import MailboxViewSet from core.api.viewsets.mailbox_access import MailboxAccessViewSet @@ -98,6 +99,9 @@ inbound_nested_router.register( r"widget", InboundWidgetViewSet, basename="inbound-widget" ) +inbound_nested_router.register( + r"brevo", InboundBrevoViewSet, basename="inbound-brevo" +) # Router for /maildomains/{maildomain_id}/message-templates/