Email verification in Django

# accounts/validators.py
import logging

import requests
from django.conf import settings
from django.core.exceptions import ValidationError

logger = logging.getLogger(__name__)

_session = requests.Session()
_session.headers["Authorization"] = f"Bearer {settings.ZAPBOUNCE_KEY}"

# catch_all and unknown are not rejections. Only a definite no blocks a signup,
# plus the disposable flag, which is a boolean that sits beside the result.


def deliverable_email(value: str) -> None:
    """Field validator. Fails OPEN: a slow verifier must not cost a signup."""
    try:
        response = _session.post(
            "https://api.zapbounce.com/v1/verify",
            json={"email": value},
            timeout=(2, 3),
        )
        response.raise_for_status()
    except requests.RequestException:
        logger.warning("zapbounce unreachable, allowing %s", value)
        return

    result = response.json()
    if result["result"] == "invalid" or result["disposable"]:
        raise ValidationError("That address will not receive mail. Check it for a typo.")


# accounts/forms.py
class SignupForm(forms.Form):
    email = forms.EmailField(validators=[deliverable_email])

# contacts/tasks.py
from celery import shared_task
from django.db import transaction

from .models import Contact, ContactList


@shared_task(bind=True, max_retries=3, retry_backoff=True, retry_jitter=True)
def verify_list(self, list_id: int) -> str:
    emails = list(
        Contact.objects.filter(list_id=list_id, email_result="")
        .values_list("email", flat=True)
        .iterator(chunk_size=5000)          # do not materialise a million rows
    )

    response = _session.post(
        "https://api.zapbounce.com/v1/batches",
        json={
            "name": f"list-{list_id}",
            "emails": emails,
            "webhook_url": settings.ZAPBOUNCE_WEBHOOK_URL,
        },
        headers={"Idempotency-Key": f"list-{list_id}-{self.request.id}"},
        timeout=(3, 60),
    )
    response.raise_for_status()

    batch_id = response.json()["batch_id"]
    with transaction.atomic():
        ContactList.objects.filter(pk=list_id).update(batch_id=batch_id)
    return batch_id


@shared_task
def import_results(batch_id: str) -> None:
    """Called from the webhook view. Bulk-updates rather than saving row by row."""
    updates = []
    for row in paginate_results(batch_id):
        updates.append(Contact(email=row["email"], email_result=row["result"]))

    Contact.objects.bulk_update(updates, ["email_result"], batch_size=1000)

# contacts/views.py
import hashlib
import hmac
import time

from django.http import HttpResponse, HttpResponseForbidden
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST


@csrf_exempt          # no session, no token: Django would 403 this before it runs
@require_POST
def zapbounce_webhook(request):
    raw = request.body                       # bytes, before any parsing
    header = request.headers.get("ZapBounce-Signature", "")
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)

    if abs(time.time() - int(parts.get("t", 0))) > 300:
        return HttpResponseForbidden("stale signature")

    expected = hmac.new(
        settings.ZAPBOUNCE_WEBHOOK_SECRET.encode(),
        f"{parts['t']}.".encode() + raw,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected, parts.get("v1", "")):
        return HttpResponseForbidden("bad signature")

    event = json.loads(raw)

    # Deliveries repeat. get_or_create makes a replay a no-op.
    _, created = WebhookEvent.objects.get_or_create(id=event["id"])
    if created:
        import_results.delay(event["data"]["batch_id"])

    return HttpResponse(status=204)