Disposable Email Detection¶

import hashlib
import json
import re
from pathlib import Path
from typing import Literal

# Load curated list (npm package ships JSON; or use python equivalent)
# pip install disposable-email-domains  (Python port: pyIsEmail / disposable-email-domains)
# For the auto-generated list, fetch:
# https://raw.githubusercontent.com/disposable/disposable/master/domains.json

CURATED: set[str] = set()   # populated from disposable-email-domains
AUTO_GEN: set[str] = set()  # populated from disposable/disposable daily

def load_lists(curated_path: str, autogen_path: str) -> None:
    global CURATED, AUTO_GEN
    with open(curated_path) as f:
        CURATED = set(json.load(f))
    with open(autogen_path) as f:
        AUTO_GEN = set(json.load(f))

def check_domain(domain: str) -> Literal["block", "softblock", "pass"]:
    d = domain.lower()
    if d in CURATED:
        return "block"
    if d in AUTO_GEN:
        return "softblock"
    return "pass"

def canonicalize_email(email: str) -> str:
    """
    Canonical form used ONLY for dedup lookups, never shown to user.
    Covers: gmail.com, googlemail.com (dots + plus-addressing).
    Other providers (Yahoo, Outlook) do NOT strip dots — do not generalize.
    """
    local, _, domain = email.lower().strip().partition("@")
    domain = domain.strip()

    # googlemail.com is an alias for gmail.com
    if domain == "googlemail.com":
        domain = "gmail.com"

    if domain == "gmail.com":
        # Strip plus-suffix: [email protected] -> [email protected]
        local = local.split("+")[0]
        # Remove dots: j.o.h.n -> john
        local = local.replace(".", "")

    return f"{local}@{domain}"

def email_hashes(raw_email: str) -> dict[str, str]:
    canonical = canonicalize_email(raw_email)
    return {
        "raw_hash": hashlib.sha256(raw_email.lower().encode()).hexdigest(),
        "canonical_hash": hashlib.sha256(canonical.encode()).hexdigest(),
        "canonical": canonical,  # store temporarily for dedup query; discard after
    }

import dns.asyncresolver  # dnspython >= 2.0

async def has_mx(domain: str) -> bool:
    """No MX -> hard-block. MX present -> not sufficient (catch-all passes). Cache 24 h."""
    try:
        answers = await dns.asyncresolver.resolve(domain, "MX")
        return len(answers) > 0
    except Exception:
        return False
# Node: dns.promises.resolveMx(domain) -> records.length > 0

# pip install minfraud
from minfraud import AsyncClient

async def minfraud_score(email_sha256: str, ip: str, fp: str | None, acct: int, key: str) -> dict:
    async with AsyncClient(account_id=acct, license_key=key) as c:
        req = {"device": {"ip_address": ip}, "email": {"address": email_sha256, "hash_address": True}}
        if fp:
            req["device"]["fingerprint"] = fp
        r = await c.score(req)
        return {
            "risk_score": r.risk_score,           # 0-100
            "is_disposable": r.email.is_disposable if r.email else None,
            "is_anonymous": r.ip_address.traits.is_anonymous if r.ip_address else None,
        }

# Redis sorted-set velocity check (Python/redis-py)
import redis.asyncio as redis
import time

r = redis.Redis()

async def fingerprint_signup_count_30d(fp: str) -> int:
    key = f"fp_signups:{fp}"
    now = time.time()
    cutoff = now - 30 * 86400
    # Remove entries older than 30 days
    await r.zremrangebyscore(key, "-inf", cutoff)
    count = await r.zcard(key)
    # Record this signup attempt
    await r.zadd(key, {str(now): now})
    await r.expire(key, 31 * 86400)
    return count

# Thresholds (tune via shadow-mode data before enforcing):
# count > 3 in 30d -> soft-block (require phone verify)
# count > 10 in 30d -> hard-block

import asyncio
import hashlib
import json
import time
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Literal

import aiohttp
import dns.asyncresolver
import redis.asyncio as aioredis

CURATED_URL = "https://cdn.jsdelivr.net/npm/disposable-email-domains@latest/disposable_email_blocklist.conf"
AUTOGEN_URL = "https://raw.githubusercontent.com/disposable/disposable/master/domains.json"

_curated: set[str] = set()
_autogen: set[str] = set()
_redis: aioredis.Redis | None = None


async def refresh_lists(session: aiohttp.ClientSession) -> None:
    global _curated, _autogen
    async with session.get(CURATED_URL) as r:
        text = await r.text()
        _curated = {line.strip() for line in text.splitlines() if line.strip()}
    async with session.get(AUTOGEN_URL) as r:
        _autogen = set(await r.json())


def canonicalize(email: str) -> str:
    local, _, domain = email.lower().strip().partition("@")
    if domain == "googlemail.com":
        domain = "gmail.com"
    if domain == "gmail.com":
        local = local.split("+")[0].replace(".", "")
    return f"{local}@{domain}"


def sha256(value: str) -> str:
    return hashlib.sha256(value.encode()).hexdigest()


async def mx_ok(domain: str, r: aioredis.Redis) -> bool:
    cache_key = f"mx:{domain}"
    cached = await r.get(cache_key)
    if cached is not None:
        return cached == b"1"
    try:
        answers = await dns.asyncresolver.resolve(domain, "MX")
        result = len(answers) > 0
    except Exception:
        result = False
    await r.setex(cache_key, 86400, b"1" if result else b"0")
    return result


@dataclass
class Decision:
    action: Literal["allow", "softblock", "block"]
    reason: str
    score: int


async def evaluate(
    raw_email: str,
    ip: str,
    fingerprint: str | None,
    r: aioredis.Redis,
) -> Decision:
    domain = raw_email.lower().split("@")[-1]

    # Blocklist
    if domain in _curated:
        return Decision("block", "disposable_curated", 100)
    if domain in _autogen:
        return Decision("softblock", "disposable_autogen", 70)

    # MX
    if not await mx_ok(domain, r):
        return Decision("block", "no_mx", 100)

    # Canonical dedup (pseudo — implement with real DB)
    canonical = canonicalize(raw_email)
    # exists = await db.fetchval("SELECT 1 FROM users WHERE email_canonical_hash=$1", sha256(canonical))
    # if exists: return Decision("block", "canonical_duplicate", 100)

    # Fingerprint velocity
    if fingerprint:
        key = f"fp:{fingerprint}"
        now = time.time()
        await r.zremrangebyscore(key, "-inf", now - 30 * 86400)
        count = await r.zcard(key)
        await r.zadd(key, {str(now): now})
        await r.expire(key, 31 * 86400)
        if count > 10:
            return Decision("block", "fp_velocity", 90)
        if count > 3:
            return Decision("softblock", "fp_velocity_warn", 65)

    # IP velocity
    ip_key = f"ip_h:{ip}"
    ip_count = int(await r.incr(ip_key))
    if ip_count == 1:
        await r.expire(ip_key, 3600)
    if ip_count > 5:
        return Decision("softblock", "ip_velocity", 60)

    return Decision("allow", "clean", 0)


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        await refresh_lists(session)

    r = aioredis.Redis()
    result = await evaluate("[email protected]", "1.2.3.4", "fp-abc123", r)
    print(result)  # Decision(action='allow', reason='clean', score=0)
    # canonical: [email protected] — dots and suffix stripped


if __name__ == "__main__":
    asyncio.run(main())