How to validate email addresses in Python

pip install dnspython disposable-email-domains

import re

# RFC 5322 "atext": the characters allowed in each dot-separated part of the local part.
ATOM = re.compile(r"[a-z0-9!#$%&'*+/=?^_`{|}~-]+", re.IGNORECASE)
# A DNS label: letters, digits and hyphens, not starting or ending with a hyphen.
LABEL = re.compile(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?")


def to_ascii_domain(domain):
    """Lower-cases a domain and converts international names to punycode, or returns None."""
    try:
        return domain.strip().rstrip(".").encode("idna").decode("ascii").lower()
    except UnicodeError:
        return None


def check_syntax(value):
    email = str(value).strip()
    if len(email) > 254:
        return {"ok": False, "reason": "too_long"}

    local, at, raw_domain = email.rpartition("@")
    if not at or not local or not raw_domain:
        return {"ok": False, "reason": "missing_at_or_part"}

    if len(local.encode()) > 64:
        return {"ok": False, "reason": "local_part_too_long"}
    # Splitting on "." also rejects leading, trailing and doubled dots (they create empty atoms).
    if not all(ATOM.fullmatch(atom) for atom in local.split(".")):
        return {"ok": False, "reason": "invalid_local_part"}

    domain = to_ascii_domain(raw_domain)
    labels = domain.split(".") if domain else []
    if len(labels) < 2 or not all(LABEL.fullmatch(label) for label in labels) or labels[-1].isdigit():
        return {"ok": False, "reason": "invalid_domain"}

    return {"ok": True, "local": local, "domain": domain, "normalized": f"{local.lower()}@{domain}"}

import dns.exception
import dns.name
import dns.resolver

# A dedicated resolver with a short time limit, so a slow DNS server can't stall a sign-up.
resolver = dns.resolver.Resolver()
resolver.lifetime = 3.0


def has_address(domain):
    for record_type in ("A", "AAAA"):
        try:
            resolver.resolve(domain, record_type)
            return True
        except dns.exception.DNSException:
            continue
    return False


def check_domain(domain):
    try:
        answer = resolver.resolve(domain, "MX")
        records = sorted(answer, key=lambda r: r.preference)
        # RFC 7505 "null MX": a single record pointing at "." means "this domain accepts no mail".
        if len(records) == 1 and records[0].exchange == dns.name.root:
            return {"status": "invalid", "reason": "domain_rejects_mail"}
        return {"status": "ok", "mx": [r.exchange.to_text(omit_final_dot=True) for r in records]}
    except dns.resolver.NXDOMAIN:
        return {"status": "invalid", "reason": "domain_not_found"}
    except dns.resolver.NoAnswer:
        pass  # The domain exists but has no MX records. Handled below.
    except dns.exception.DNSException:
        # SERVFAIL, timeouts and other DNS trouble are temporary. Don't blame the user.
        return {"status": "unknown", "reason": "dns_error"}

    # No MX records: mail servers fall back to the domain's own A/AAAA record ("implicit MX").
    if has_address(domain):
        return {"status": "risky", "reason": "no_mx_records"}
    return {"status": "invalid", "reason": "no_mail_server"}

import time

from dns_check import check_domain

TTL_SECONDS = 3600
_cache = {}


def check_domain_cached(domain):
    hit = _cache.get(domain)
    if hit and hit[0] > time.monotonic():
        return hit[1]

    result = check_domain(domain)
    # Don't remember temporary failures.
    if result["status"] != "unknown":
        _cache[domain] = (time.monotonic() + TTL_SECONDS, result)
    return result

from disposable_email_domains import blocklist  # a set of domain names


def is_disposable(domain):
    """Checks the domain and each parent, so "x.mailinator.com" matches "mailinator.com"."""
    labels = domain.split(".")
    return any(".".join(labels[i:]) in blocklist for i in range(len(labels) - 1))

ROLE_ACCOUNTS = {
    "admin", "billing", "contact", "help", "hello", "info", "marketing", "no-reply",
    "noreply", "office", "postmaster", "sales", "support", "team", "webmaster",
}


def is_role_account(local):
    # Ignore "+tag" sub-addressing: "support+billing" is still "support".
    return local.lower().split("+")[0] in ROLE_ACCOUNTS


# Real providers people mistype. Include close neighbours (mail.com, ymail.com, gmx.com…)
# too, or they'd be "corrected" to the provider one letter away.
POPULAR_DOMAINS = [
    "gmail.com", "googlemail.com", "yahoo.com", "ymail.com", "yahoo.co.uk", "hotmail.com",
    "hotmail.co.uk", "outlook.com", "live.com", "msn.com", "icloud.com", "me.com", "mac.com",
    "aol.com", "mail.com", "gmx.com", "gmx.de", "proton.me", "protonmail.com",
]


def distance(a, b):
    """Levenshtein distance where swapping two adjacent letters counts as one edit."""
    d = [[i + j if i == 0 or j == 0 else 0 for j in range(len(b) + 1)] for i in range(len(a) + 1)]
    for i in range(1, len(a) + 1):
        for j in range(1, len(b) + 1):
            cost = 0 if a[i - 1] == b[j - 1] else 1
            d[i][j] = min(d[i - 1][j] + 1, d[i][j - 1] + 1, d[i - 1][j - 1] + cost)
            if i > 1 and j > 1 and a[i - 1] == b[j - 2] and a[i - 2] == b[j - 1]:
                d[i][j] = min(d[i][j], d[i - 2][j - 2] + 1)
    return d[len(a)][len(b)]


def suggest_domain(domain):
    """"gmial.com" -> "gmail.com"; None when the domain is fine or nothing is close."""
    if domain in POPULAR_DOMAINS:
        return None
    best, best_distance = min(((c, distance(domain, c)) for c in POPULAR_DOMAINS), key=lambda x: x[1])
    return best if best_distance <= (1 if len(domain) <= 6 else 2) else None

from disposable import is_disposable
from dns_cache import check_domain_cached
from extras import is_role_account, suggest_domain
from syntax import check_syntax


def validate_email(value):
    """Returns {"status", "reason", "email", "did_you_mean"}.

    status is "valid", "invalid", "risky" or "unknown".
    """
    syntax = check_syntax(value)
    if not syntax["ok"]:
        return {"status": "invalid", "reason": syntax["reason"], "email": None, "did_you_mean": None}

    local, domain = syntax["local"], syntax["domain"]
    suggestion = suggest_domain(domain)
    base = {"email": syntax["normalized"], "did_you_mean": f"{local}@{suggestion}" if suggestion else None}

    dns_result = check_domain_cached(domain)
    if dns_result["status"] in ("invalid", "unknown"):
        return {**base, "status": dns_result["status"], "reason": dns_result["reason"]}

    # The domain can receive mail. Now look for risk signals, most serious first.
    if is_disposable(domain):
        return {**base, "status": "risky", "reason": "disposable"}
    if suggestion:
        return {**base, "status": "risky", "reason": "possible_typo"}
    if dns_result["status"] == "risky":
        return {**base, "status": "risky", "reason": dns_result["reason"]}
    if is_role_account(local):
        return {**base, "status": "risky", "reason": "role_based"}
    return {**base, "status": "valid", "reason": "mail_server_found"}

from flask import Flask, jsonify, request

from validation import validate_email

app = Flask(__name__)


@app.post("/signup")
def signup():
    payload = request.get_json(silent=True) or {}
    result = validate_email(payload.get("email", ""))

    if result["status"] == "invalid" or result["reason"] == "disposable":
        if result["reason"] == "disposable":
            message = "Please use a permanent email address."
        else:
            message = "That email address can't receive mail."
        return jsonify(error=message, did_you_mean=result["did_you_mean"]), 422

    # "unknown" means DNS had a hiccup. Accept the address and check it again later.
    # create_user(result["email"])
    return jsonify(ok=True, email=result["email"], did_you_mean=result["did_you_mean"]), 201

from django import forms

from validation import validate_email


class SignupForm(forms.Form):
    email = forms.EmailField()

    def clean_email(self):
        result = validate_email(self.cleaned_data["email"])
        if result["status"] == "invalid":
            raise forms.ValidationError("That email address can't receive mail.")
        if result["reason"] == "disposable":
            raise forms.ValidationError("Please use a permanent email address.")
        return result["email"]

import pytest

from syntax import check_syntax


@pytest.mark.parametrize("email", ["jane@example.com", "jane.doe+news@example.co.uk", "user@münchen.de"])
def test_accepts_normal_addresses(email):
    assert check_syntax(email)["ok"]


@pytest.mark.parametrize(
    "email", ["", "jane", "@example.com", "jane@", "a..b@example.com", "jane@example", "jane@-example.com"]
)
def test_rejects_malformed_addresses(email):
    assert not check_syntax(email)["ok"]

import logging
import os

import requests

API = "https://validateemail.io/v1"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['VALIDATEEMAIL_KEY']}"


def validate_with_api(email):
    try:
        res = session.get(f"{API}/validate", params={"email": email}, timeout=5)
        res.raise_for_status()
        return res.json()
    except requests.RequestException as err:
        # Never block a sign-up because the validator is slow or unreachable.
        logging.warning("Email validation unavailable: %s", err)
        return {"status": "unknown", "reason": "validator_unavailable"}


def validate_list(emails):
    """Validates any number of addresses, 100 per request."""
    results = []
    for i in range(0, len(emails), 100):
        res = session.post(f"{API}/validate/batch", json={"emails": emails[i : i + 100]}, timeout=30)
        res.raise_for_status()
        results.extend(res.json()["results"])
    return results