import asyncio import re import time import uuid from datetime import datetime, timezone from typing import Any, Optional import structlog from ..models import ( ValidationResult, ValidationStatus, ScoringResult, CatchAllResult, DNSResult, DisposableResult, SMTPResult, SyntaxResult, ) from ..config import settings from .syntax import validate_syntax from .dns_check import check_dns from .disposable import check_disposable, domain_in_blocklist, domain_is_known from .mailbox_provider import MailboxProvider, resolve_provider from .i18n import EAI_REASON, eai_gate from ..ml.risk_scorer import AIRiskScorer, ai_scorer, label_for_score, score_many_by_domain from ..ml.runtime import get_model_async, predict_async from .mx_intel import mx_is_trusted from .retry_queue import is_greylist_result from .scoring import calculate_score logger = structlog.get_logger() def _mask_email_for_log(email: str) -> str: local_part, separator, domain = email.rpartition("@") if not separator or not domain: return "[invalid-email]" return f"{local_part[:1]}***@{domain}" def _build_checks_run( syntax: SyntaxResult | None, dns: DNSResult | None, disposable: DisposableResult | None, catch_all: CatchAllResult | None, smtp: SMTPResult | None, scoring: ScoringResult | None, ) -> list[str]: """Which of the 7-layer taxonomy keys actually executed for this result. Mirrors the frontend fallback in src/lib/verifier.ts classify() so a layer the engine did not run is never reported as a pass. DNS without an error claims both `dns` and `mx`; a DNS error claims only `dns`. """ checks: list[str] = [] if syntax is not None: checks.append("syntax") if dns is not None: if dns.error: checks.append("dns") else: checks.extend(["dns", "mx"]) if disposable is not None: checks.append("disposable") if catch_all is not None: checks.append("catch_all") if smtp is not None: checks.append("smtp") if scoring is not None: checks.append("scoring") return checks def _build_reason_code( dns: DNSResult | None, disposable: DisposableResult | None, catch_all: CatchAllResult | None, smtp: SMTPResult | None, *, early: str | None = None, ) -> str | None: """Machine cause for the verdict. Ordered most-specific/actionable first. disposable outranks everything (frontend classifies it as Disposable). A definitive SMTP rejection (verified False + code) outranks the domain-level catch-all flag: reporting "catch_all_detected" beside status=invalid is incoherent — the explicit "mailbox does not exist" is the actionable fact. Catch-all then outranks a merely inconclusive probe. """ if early: return early if disposable is not None and disposable.is_disposable: return "disposable" if dns is not None and dns.a_fallback and not dns.has_mx_records: return "a_fallback_only" if smtp is not None and smtp.verified is False and smtp.smtp_code: if smtp.smtp_code >= 500: return "rcpt_5xx" if smtp.smtp_code >= 400: return "rcpt_4xx" if smtp.smtp_code >= 200: return "rcpt_2xx" if catch_all is not None and catch_all.is_catch_all: return "catch_all_detected" if smtp is not None and smtp.verified is None and smtp.error: return ( "smtp_timeout" if re.search(r"timed?\s*out", smtp.error, re.IGNORECASE) else "smtp_unavailable" ) return None def _deterministic_verdict( disposable: DisposableResult | None, catch_all: CatchAllResult | None, smtp: SMTPResult | None, ) -> str: """Pre-AI verdict of the deterministic gauntlet (spec Phase 5). Order matters: an explicit mailbox rejection outranks the domain-level catch-all flag (mirrors _build_reason_code), and a catch-all NEVER reads "valid" even when the probe was accepted — it only proves the domain accepts everything. """ if disposable is not None and disposable.is_disposable: return "disposable" if ( smtp is not None and smtp.verified is False and smtp.smtp_code is not None and smtp.smtp_code >= 500 ): return "invalid" if catch_all is not None and catch_all.is_catch_all: return "catch_all" if smtp is not None and smtp.verified is True: return "valid" return "unknown" def _smtp_class(smtp: SMTPResult | None) -> str: """Coarse SMTP response class for the AI feature string (no PII).""" if smtp is None: return "skipped" if smtp.verified is True: return "accept" if smtp.verified is False: if smtp.smtp_code and smtp.smtp_code >= 500: return "5xx" if smtp.smtp_code and smtp.smtp_code >= 400: return "4xx" return "reject" err = (smtp.error or "").lower() if "worker" in err or "unreachable" in err: return "transport_error" return "unavailable" def _ambiguous(verdict: str) -> bool: """Only these two reach the AI scorer (spec gate steps 6 and 8).""" return verdict in {"catch_all", "unknown"} class EmailValidationEngine: """ Master 7-Layer Email Validation Orchestrator. Layers: 1. Syntax / RFC 5321 check 2. DNS domain existence 3. MX record verification 4. Disposable / Role-based / Free provider detection 5. Catch-all domain detection 6. SMTP mailbox verification 7. Deterministic heuristic deliverability scoring """ def __init__( self, mailbox: MailboxProvider | None = None, cache=None, retry_queue=None, scorer: AIRiskScorer | None = None, ): self.use_proxies = settings.use_proxies # Layers 5 (catch-all) and 6 (SMTP) run through the autonomous mailbox # layer: the built-in SMTP engine on this host, or the remote SMTP # probe worker when SMTP_WORKER_URL is configured (spec §3.3). self.mailbox = mailbox if mailbox is not None else resolve_provider() # DNS/MX and definitive mailbox verdicts are cached across runs — the # same domain (and often the same address) recurs constantly in bulk # lists. Transient failures are never cached; see validation_cache.py. if cache is None: from .validation_cache import validation_cache as cache self.cache = cache # Greylisted (4xx) addresses get a bounded second chance. Without this a # valid mailbox on a greylisting domain stays permanently "risky". if retry_queue is None: from .retry_queue import retry_queue as retry_queue self.retry_queue = retry_queue # AI risk scorer (spec §3.4): pattern-based classification for the # ambiguous rows only; circuit breaker + heuristic fallback inside. self.scorer = scorer if scorer is not None else ai_scorer async def validate(self, email: str, deep: bool = True) -> ValidationResult: start_time = time.monotonic() run_id = uuid.uuid4().hex[:12] log = logger.bind(email=_mask_email_for_log(email), run_id=run_id) log.info("validation_started") # ---- Layer 1: Syntax ----------------------------------------- syntax = validate_syntax(email) if not syntax.passed: elapsed = (time.monotonic() - start_time) * 1000 scoring = calculate_score(syntax, None, None, None, None) log.warning("syntax_failed", error=syntax.error) return ValidationResult( email=email, status=ValidationStatus.INVALID, syntax=syntax, scoring=scoring, processing_time_ms=round(elapsed, 2), validated_at=datetime.now(timezone.utc).isoformat(), checks_run=_build_checks_run(syntax, None, None, None, None, scoring), reason_code=_build_reason_code(None, None, None, None), run_id=run_id, validator_version=settings.app_version, deterministic_verdict="invalid", ) domain = syntax.domain normalized_email = syntax.normalized_email or email # ---- Layers 2 & 3: DNS + MX --------------------------------- dns_result = await check_dns(domain, settings.dns_timeout) if dns_result.error_code in {"timeout", "resolver_error"}: elapsed = (time.monotonic() - start_time) * 1000 scoring = calculate_score(syntax, dns_result, None, None, None) log.warning( "dns_inconclusive", domain=domain, error_code=dns_result.error_code ) return ValidationResult( email=normalized_email, status=self._determine_status(scoring), syntax=syntax, dns=dns_result, scoring=scoring, processing_time_ms=round(elapsed, 2), validated_at=datetime.now(timezone.utc).isoformat(), checks_run=_build_checks_run(syntax, dns_result, None, None, None, scoring), reason_code=_build_reason_code(dns_result, None, None, None, early="dns_timeout"), run_id=run_id, validator_version=settings.app_version, deterministic_verdict="unknown", ) if ( not dns_result.domain_exists or dns_result.error_code == "nxdomain" or (not dns_result.has_mx_records and not dns_result.a_fallback) ): elapsed = (time.monotonic() - start_time) * 1000 scoring = calculate_score(syntax, dns_result, None, None, None) log.warning("dns_failed", domain=domain) return ValidationResult( email=normalized_email, status=ValidationStatus.INVALID, syntax=syntax, dns=dns_result, scoring=scoring, processing_time_ms=round(elapsed, 2), validated_at=datetime.now(timezone.utc).isoformat(), checks_run=_build_checks_run(syntax, dns_result, None, None, None, scoring), reason_code=_build_reason_code( dns_result, None, None, None, early="nxdomain" ), run_id=run_id, validator_version=settings.app_version, deterministic_verdict="invalid", ) # ---- Layer 4: Disposable Detection --------------------------- disposable = check_disposable(normalized_email, domain) # ML advisory (single path): unknown domains only, fail-open. Mirrors # the batch pipeline so both surfaces agree on the same address. # Skipped when the MX runs on a trusted enterprise platform: the model # learned 'unfamiliar domain string = disposable' from an imbalanced # dataset and flags legitimate small businesses on Google Workspace / # M365 — the deterministic MX intelligence is the better signal there. ml_model = await get_model_async() mx_host = ( dns_result.mx_records[0]["host"] if dns_result.mx_records else None ) if ( ml_model is not None and not domain_is_known(domain) and deep and not mx_is_trusted(mx_host) ): # Off the event loop: CPU inference inline would stall every other # concurrent request (measured 42.6s cold start before this). preds = await predict_async(ml_model, [f"john.smith [SEP] {domain}"]) if preds: disposable = disposable.model_copy( update={"ml_label": preds[0][0], "ml_confidence": preds[0][1]} ) if not deep: # Fast path: skip SMTP for quick validation elapsed = (time.monotonic() - start_time) * 1000 scoring = calculate_score( syntax, dns_result, disposable, None, None, mailbox_verification_skipped=True, ) status = self._determine_status(scoring) return ValidationResult( email=normalized_email, status=status, syntax=syntax, dns=dns_result, disposable=disposable, scoring=scoring, processing_time_ms=round(elapsed, 2), validated_at=datetime.now(timezone.utc).isoformat(), checks_run=_build_checks_run( syntax, dns_result, disposable, None, None, scoring ), reason_code=_build_reason_code(dns_result, disposable, None, None), run_id=run_id, validator_version=settings.app_version, # Fast path: the mailbox question was never asked, so there is # no deterministic mailbox verdict to report. deterministic_verdict=None, ) # ---- Layers 5 & 6: Catch-All + SMTP -------------------------- mx_host = dns_result.mx_records[0]["host"] # EAI/SMTPUTF8 gate (spec §7.2): a non-ASCII local part can only be # delivered when the receiving MX advertises SMTPUTF8. Without an SMTP # dialogue the engine refuses the per-address probe deterministically # instead of faking a verdict. The domain-level catch-all probe (a # random ASCII local part) still runs — it is independent of the # address's own local part. eai_reason = eai_gate(syntax.local_part or "") catch_all_task = asyncio.create_task( self.mailbox.catch_all(domain, mx_host) ) if eai_reason: smtp_result = SMTPResult( verified=None, error=( f"{EAI_REASON}: mailbox probe requires SMTPUTF8 support " "(deterministic refusal)" ), ) try: catch_all_raw = await catch_all_task except Exception as exc: # Same contract as the gather path: a failed probe is UNKNOWN. catch_all_raw = exc else: smtp_task = asyncio.create_task( self.mailbox.verify(normalized_email, mx_host) ) catch_all_raw, smtp_result = await asyncio.gather( catch_all_task, smtp_task, return_exceptions=True ) # Handle catch-all result. An exception means the probe could not run # (e.g. port-25 egress blocked) — that is UNKNOWN, not "not catch-all". if isinstance(catch_all_raw, Exception): catch_all = CatchAllResult( is_catch_all=None, confidence=0.0, error=str(catch_all_raw) ) else: is_catch_all, confidence = catch_all_raw catch_all = CatchAllResult(is_catch_all=is_catch_all, confidence=confidence) # Handle SMTP result if isinstance(smtp_result, Exception): smtp_result = SMTPResult(verified=None, error=str(smtp_result)) # ---- Layer 7: Scoring ---------------------------------------- scoring = calculate_score( syntax, dns_result, disposable, catch_all, smtp_result ) status = self._determine_status(scoring) elapsed = (time.monotonic() - start_time) * 1000 # ---- Normalized verdict (spec Phase 5): deterministic first, then AI -- deterministic = _deterministic_verdict(disposable, catch_all, smtp_result) ai_score: float | None = None ai_label: str | None = None fallback_used = False if deep and _ambiguous(deterministic): features = AIRiskScorer.build_features( domain=domain, mx_host=mx_host, smtp_class=_smtp_class(smtp_result), catch_all=catch_all.is_catch_all if catch_all else None, local_part=syntax.local_part or "", dns_inconsistent=dns_result.dns_inconsistent, ) scored = await self.scorer.score(features) ai_score, ai_label, fallback_used = ( scored.score, scored.label, scored.fallback_used, ) if scored.label == "high_risk": scoring.warnings.append("ai_high_risk") if dns_result.dns_inconsistent: scoring.warnings.append("dns_inconsistent: resolvers disagree on MX") reason = _build_reason_code(dns_result, disposable, catch_all, smtp_result) if eai_reason: reason = EAI_REASON log.info( "validation_complete", status=status, score=scoring.score, time_ms=round(elapsed, 2), ) return ValidationResult( email=normalized_email, status=status, syntax=syntax, dns=dns_result, disposable=disposable, catch_all=catch_all, smtp=smtp_result, scoring=scoring, processing_time_ms=round(elapsed, 2), validated_at=datetime.now(timezone.utc).isoformat(), checks_run=_build_checks_run( syntax, dns_result, disposable, catch_all, smtp_result, scoring ), reason_code=reason, run_id=run_id, validator_version=settings.app_version, deterministic_verdict=deterministic, ai_risk_score=ai_score, ai_label=ai_label, fallback_used=fallback_used, ) async def validate_batch( self, emails: list[str], deep: bool = True ) -> list[ValidationResult]: """Validate many addresses in one pass, reusing per-domain work. Bulk throughput comes from doing DNS/MX **once per domain** and handing every (email, mx_host) pair to the mailbox provider in a **single** call, so the SMTP engine can reuse one session per receiving host instead of a fresh TCP handshake per address. Providers that cannot batch fall back to the per-address path, so behaviour is never worse than before. Results are returned in the same order as `emails`. """ if not emails: return [] run_id = uuid.uuid4().hex[:12] out: list[ValidationResult | None] = [None] * len(emails) def _stamp(**kw) -> ValidationResult: return ValidationResult( processing_time_ms=0.0, validated_at=datetime.now(timezone.utc).isoformat(), run_id=run_id, validator_version=settings.app_version, **kw, ) # ---- Layer 1: syntax (per address) --------------------------- parsed: list[tuple[str, SyntaxResult]] = [] domain_index: dict[str, list[int]] = {} for idx, raw in enumerate(emails): syntax = validate_syntax(raw) parsed.append((raw, syntax)) if not syntax.passed: scoring = calculate_score(syntax, None, None, None, None) out[idx] = _stamp( email=raw, status=ValidationStatus.INVALID, syntax=syntax, scoring=scoring, checks_run=_build_checks_run(syntax, None, None, None, None, scoring), reason_code=_build_reason_code(None, None, None, None), ) continue domain_index.setdefault(syntax.domain, []).append(idx) # ---- Layers 2 & 3: DNS + MX once per domain (cached) --------- domains = list(domain_index.keys()) domain_dns: dict[str, DNSResult] = {} misses: list[str] = [] for domain in domains: cached_dns = await self.cache.get_dns(domain) if cached_dns is not None: domain_dns[domain] = cached_dns else: misses.append(domain) if misses: resolved = await asyncio.gather( *[check_dns(d, settings.dns_timeout) for d in misses], return_exceptions=True, ) for domain, dns in zip(misses, resolved): result = ( dns if isinstance(dns, DNSResult) else DNSResult( domain_exists=False, has_mx_records=False, error=str(dns), error_code="resolver_error", ) ) domain_dns[domain] = result # put_dns ignores transient failures internally. await self.cache.put_dns(domain, result) pending: list[tuple[int, str, SyntaxResult, str, DNSResult]] = [] for domain, idxs in domain_index.items(): dns = domain_dns[domain] inconclusive = dns.error_code in {"timeout", "resolver_error"} dead = ( not dns.domain_exists or dns.error_code == "nxdomain" or (not dns.has_mx_records and not dns.a_fallback) ) for idx in idxs: raw, syntax = parsed[idx] normalized = syntax.normalized_email or raw if inconclusive: scoring = calculate_score(syntax, dns, None, None, None) out[idx] = _stamp( email=normalized, status=self._determine_status(scoring), syntax=syntax, dns=dns, scoring=scoring, checks_run=_build_checks_run( syntax, dns, None, None, None, scoring ), reason_code=_build_reason_code( dns, None, None, None, early="dns_timeout" ), deterministic_verdict="unknown", ) elif dead: scoring = calculate_score(syntax, dns, None, None, None) out[idx] = _stamp( email=normalized, status=ValidationStatus.INVALID, syntax=syntax, dns=dns, scoring=scoring, checks_run=_build_checks_run( syntax, dns, None, None, None, scoring ), reason_code=_build_reason_code( dns, None, None, None, early="nxdomain" ), deterministic_verdict="invalid", ) else: pending.append((idx, normalized, syntax, domain, dns)) # ---- Layer 4 fast path (no mailbox probe) -------------------- if pending and not deep: for idx, normalized, syntax, domain, dns in pending: disposable = check_disposable(normalized, domain) scoring = calculate_score( syntax, dns, disposable, None, None, mailbox_verification_skipped=True, ) out[idx] = _stamp( email=normalized, status=self._determine_status(scoring), syntax=syntax, dns=dns, disposable=disposable, scoring=scoring, checks_run=_build_checks_run( syntax, dns, disposable, None, None, scoring ), reason_code=_build_reason_code(dns, disposable, None, None), deterministic_verdict=None, ) pending = [] # ---- Layers 5 & 6: ONE batched mailbox call ------------------ cached_verdicts: dict[str, tuple[SMTPResult, Optional[bool]]] = {} catch_all_map: dict[str, Optional[bool]] = {} if pending: # Serve anything already answered (definitive verdicts only) so we # never re-probe an address we already know the answer to. to_probe: list[tuple[int, str, SyntaxResult, str, DNSResult]] = [] for item in pending: _idx, normalized, _syntax, _domain, _dns = item hit = await self.cache.get_verdict(normalized, deep) if hit is None: to_probe.append(item) else: cached_verdicts[normalized] = hit if hit[1] is not None: catch_all_map[item[3]] = hit[1] smtp_map: dict[str, SMTPResult] = {} pairs: list[tuple[str, str]] = [] for _idx, normalized, _syntax, _domain, _dns in to_probe: if eai_gate(_syntax.local_part or ""): # Deterministic EAI refusal (spec §7.2): the probe requires # an SMTPUTF8 advertisement we cannot observe, so refuse # instead of guessing. Cache/check semantics unchanged. smtp_map[normalized] = SMTPResult( verified=None, error=( f"{EAI_REASON}: mailbox probe requires SMTPUTF8 " "(deterministic refusal)" ), ) continue if _dns.mx_records: pairs.append((normalized, _dns.mx_records[0]["host"])) else: smtp_map[normalized] = SMTPResult( verified=None, error="no MX host available for probe" ) if pairs: try: if hasattr(self.mailbox, "verify_many"): smtp_map, caught = await self.mailbox.verify_many(pairs) for domain, value in (caught or {}).items(): catch_all_map.setdefault(domain, value) else: # Provider cannot batch: per-address fallback. for email, mx in pairs: smtp_map[email] = await self.mailbox.verify(email, mx) for domain in {e.rsplit("@", 1)[-1].lower() for e, _ in pairs}: mx = next( m for e, m in pairs if e.rsplit("@", 1)[-1].lower() == domain ) value, _conf = await self.mailbox.catch_all(domain, mx) catch_all_map.setdefault(domain, value) except Exception as exc: # one bad stage must not abort the batch smtp_map = { e: SMTPResult(verified=None, error=f"mailbox_stage_error: {exc}") for e, _ in pairs } # ---- ML advisory signal (unknown domains, ONE batched call) ---- # Runs only where the curated lists have no answer - the exact gap # a static blocklist cannot close. Any failure here is silently # ignored: the deterministic verdict stands. # Domains on trusted enterprise MX (Google Workspace / M365 / # gateways) are excluded: disposable domains essentially never run # there, and the model's 'unfamiliar string' bias would otherwise # flag legitimate businesses as disposable. ml_map: dict[str, tuple[str, float]] = {} ml_model = await get_model_async() if ml_model is not None: unknown_domains = sorted( { domain for _, _, _, domain, dns in pending if not domain_is_known(domain) and dns.mx_records and not mx_is_trusted(dns.mx_records[0]["host"]) } ) if unknown_domains: # One batched call, off the event loop (CPU inference inline # stalls concurrent requests), and fail-open by construction: # predict_async never raises. preds = await predict_async( ml_model, [f"john.smith [SEP] {d}" for d in unknown_domains] ) if preds: ml_map = dict(zip(unknown_domains, preds)) # ---- Layer 7: score each row ----------------------------- ai_cache: dict[str, Any] = {} # feature string -> AIScore (dedupe) for idx, normalized, syntax, domain, dns in pending: cached = cached_verdicts.get(normalized) if cached is not None: smtp, ca_cached = cached else: smtp = smtp_map.get(normalized) or SMTPResult( verified=None, error="engine returned no result" ) ca_cached = None # Persist only definitive answers (verified verdict or a # confirmed catch-all); transient failures are never cached. await self.cache.put_verdict( normalized, deep, smtp, catch_all_map.get(domain) ) ca_value = ( ca_cached if ca_cached is not None else catch_all_map.get(domain) ) catch_all = CatchAllResult( is_catch_all=ca_value, confidence=0.9 if ca_value is not None else 0.0, ) disposable = check_disposable(normalized, domain) ml = ml_map.get(domain) if ml is not None: disposable = disposable.model_copy( update={"ml_label": ml[0], "ml_confidence": ml[1]} ) # A 4xx means "come back later", not a verdict. Schedule one # bounded retry so a real mailbox on a greylisting domain is not # stranded as permanently risky. The queue never raises. if cached is None: try: if is_greylist_result(smtp): await self.retry_queue.enqueue( normalized, dns.mx_records[0]["host"], smtp.smtp_code ) elif smtp.verified is not None: # Definitive answer — cancel any pending retry. await self.retry_queue.resolve_email(normalized) except Exception: pass scoring = calculate_score( syntax, dns, disposable, catch_all, smtp ) # ---- Normalized verdict (spec Phase 5) ------------------- deterministic = _deterministic_verdict(disposable, catch_all, smtp) ai_score: float | None = None ai_label: str | None = None fallback_used = False if deep and _ambiguous(deterministic): feats = AIRiskScorer.build_features( domain=domain, mx_host=dns.mx_records[0]["host"] if dns.mx_records else None, smtp_class=_smtp_class(smtp), catch_all=ca_value, local_part=syntax.local_part or "", dns_inconsistent=dns.dns_inconsistent, ) if feats not in ai_cache: ai_cache[feats] = await self.scorer.score(feats) scored = ai_cache[feats] ai_score, ai_label, fallback_used = ( scored.score, scored.label, scored.fallback_used, ) if scored.label == "high_risk": scoring.warnings.append("ai_high_risk") if dns.dns_inconsistent: scoring.warnings.append("dns_inconsistent: resolvers disagree on MX") reason = _build_reason_code(dns, disposable, catch_all, smtp) if (syntax.local_part or "") and eai_gate(syntax.local_part or ""): reason = EAI_REASON out[idx] = _stamp( email=normalized, status=self._determine_status(scoring), syntax=syntax, dns=dns, disposable=disposable, catch_all=catch_all, smtp=smtp, scoring=scoring, checks_run=_build_checks_run( syntax, dns, disposable, catch_all, smtp, scoring ), reason_code=reason, deterministic_verdict=deterministic, ai_risk_score=ai_score, ai_label=ai_label, fallback_used=fallback_used, cache_hit=cached is not None, ) # Guarantee one result per input, in order. for idx, result in enumerate(out): if result is None: raw = emails[idx] syntax = parsed[idx][1] if idx < len(parsed) else validate_syntax(raw) scoring = calculate_score(syntax, None, None, None, None) out[idx] = _stamp( email=raw, status=ValidationStatus.UNKNOWN, syntax=syntax, scoring=scoring, checks_run=_build_checks_run( syntax, None, None, None, None, scoring ), reason_code="engine_missing_result", deterministic_verdict="unknown", ) return [r for r in out if r is not None] def _determine_status(self, scoring: ScoringResult) -> ValidationStatus: if scoring.score >= 75: return ValidationStatus.VALID elif scoring.score >= 40: return ValidationStatus.RISKY else: return ValidationStatus.INVALID # Singleton engine engine = EmailValidationEngine()