FortyDN Dev
v3.0.0: fully autonomous engine — remove MillionVerifier, add DoH DNS worker, AI risk scorer (HF PRO, circuit breaker), i18n, normalized verdicts, remote SMTP worker
de745bf | """DNS/MX worker over DNS-over-HTTPS (DoH) — free public resolvers. | |
| Architecture spec §3.2 / §7.3: | |
| - MX resolution runs on Google Public DNS and Cloudflare DNS over HTTPS. | |
| Both are free, unlimited and need no authentication; HTTPS/443 egress is | |
| never blocked (unlike UDP/53 on locked-down PaaS hosts). | |
| - The SAME query goes to BOTH resolvers. Divergent MX sets flag the domain as | |
| `dns_inconsistent` — a deterministic risk signal (regional DNS hijack / | |
| misconfiguration), requiring no inference. | |
| - Results are cached at the domain level: 24h TTL for MX answers, 1h for | |
| negative (null-MX / NXDOMAIN) answers per the spec. | |
| - Punycode normalization happens here (§7.1) so Unicode domains resolve. | |
| The legacy dnspython path in dns_check.py remains the fallback: any DoH | |
| failure (network, JSON, schema) degrades to it transparently. The DoH layer | |
| can also be disabled entirely with DOH_ENABLED=false. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import json | |
| from typing import Any, Optional | |
| import httpx | |
| from ..config import settings | |
| from .i18n import to_punycode | |
| GOOGLE_DOH = "https://dns.google/resolve" | |
| CLOUDFLARE_DOH = "https://cloudflare-dns.com/dns-query" | |
| MX_TTL_SECONDS = 24 * 3600 # spec: 24h for MX records | |
| NEGATIVE_TTL_SECONDS = 1 * 3600 # spec: 1h for null-MX negatives | |
| _TIMEOUT_LABEL = "timeout" | |
| async def _doh_query( | |
| client: httpx.AsyncClient, | |
| endpoint: str, | |
| name: str, | |
| rtype: str, | |
| extra_params: dict | None = None, | |
| accept: str = "application/dns-json", | |
| ) -> dict: | |
| """One DoH JSON query (RFC 8484-style JSON APIs used by both providers).""" | |
| params = {"name": name, "type": rtype} | |
| if extra_params: | |
| params.update(extra_params) | |
| resp = await client.get(endpoint, params=params, headers={"accept": accept}) | |
| resp.raise_for_status() | |
| data = resp.json() | |
| if not isinstance(data, dict): | |
| raise ValueError("doh: non-object response") | |
| return data | |
| def _parse_mx(data: dict) -> list[dict]: | |
| """Extract MX records (sorted by priority) from a DoH answer.""" | |
| records: list[dict] = [] | |
| for ans in data.get("Answer") or []: | |
| # RFC 1035 type 15 = MX | |
| if ans.get("type") != 15: | |
| continue | |
| parts = str(ans.get("data", "")).split() | |
| if len(parts) >= 2: | |
| try: | |
| priority = int(parts[0]) | |
| except ValueError: | |
| priority = 0 | |
| host = " ".join(parts[1:]).rstrip(".").strip().lower() | |
| if host: | |
| records.append({"host": host, "priority": priority}) | |
| records.sort(key=lambda r: (r["priority"], r["host"])) | |
| return records | |
| def _parse_a_like(data: dict, rtypes: set[int]) -> bool: | |
| """True when the answer holds at least one A (1) / AAAA (28) record.""" | |
| for ans in data.get("Answer") or []: | |
| if ans.get("type") in rtypes: | |
| return True | |
| return False | |
| def _mx_sets_match(a: list[dict], b: list[dict]) -> bool: | |
| """Set-equality on (priority, host) — order-insensitive, tolerant.""" | |
| norm = lambda rs: {(r["host"], r["priority"]) for r in rs} | |
| return norm(a) == norm(b) | |
| async def resolve_mx_dual( | |
| domain: str, timeout: Optional[float] = None | |
| ) -> dict[str, Any]: | |
| """Query BOTH public resolvers for MX and compare. | |
| Returns: | |
| { | |
| "status": "ok" | "nxdomain" | "no_mail" | "error", | |
| "mx_records": [...], # primary (Google) answer | |
| "consistent": bool | None, # None when the second resolver failed | |
| "a_fallback": bool, | |
| "error": str | None, | |
| } | |
| """ | |
| domain = to_punycode((domain or "").strip().lower().rstrip(".")) | |
| timeout = timeout if timeout is not None else settings.doh_timeout | |
| async def _query_both() -> tuple[dict | None, dict | None, Exception | None]: | |
| async with httpx.AsyncClient(timeout=timeout) as client: | |
| google_task = _doh_query(client, GOOGLE_DOH, domain, "MX") | |
| cf_task = _doh_query( | |
| client, CLOUDFLARE_DOH, domain, "MX", extra_params={"ct": "application/dns-json"} | |
| ) | |
| g_res = await asyncio.gather(google_task, cf_task, return_exceptions=True) | |
| g, c = g_res | |
| g_err = g if isinstance(g, Exception) else None | |
| c_err = c if isinstance(c, Exception) else None | |
| g_data = g if isinstance(g, dict) else None | |
| c_data = c if isinstance(c, dict) else None | |
| # If the PRIMARY (Google) failed but Cloudflare answered, swap so the | |
| # pipeline still gets an authoritative answer from a public resolver. | |
| if g_data is None and c_data is not None: | |
| g_data, c_data = c_data, None | |
| return g_data, c_data, g_err or c_err | |
| try: | |
| g_data, c_data, err = await _query_both() | |
| except Exception as exc: # httpx transport failure etc. | |
| return {"status": "error", "mx_records": [], "consistent": None, | |
| "a_fallback": False, "error": f"doh transport: {exc}"} | |
| if g_data is None: | |
| return {"status": "error", "mx_records": [], "consistent": None, | |
| "a_fallback": False, "error": f"doh unavailable: {err}"} | |
| g_status = g_data.get("Status", 0) | |
| if g_status == 3: # NXDOMAIN | |
| return {"status": "nxdomain", "mx_records": [], "consistent": None, | |
| "a_fallback": False, "error": "Domain does not exist (NXDOMAIN)"} | |
| if g_status not in (0, 3): | |
| return {"status": "error", "mx_records": [], "consistent": None, | |
| "a_fallback": False, "error": f"doh resolver status {g_status}"} | |
| mx = _parse_mx(g_data) | |
| if mx: | |
| consistent = ( | |
| _mx_sets_match(mx, _parse_mx(c_data)) | |
| if c_data is not None and c_data.get("Status", 0) == 0 | |
| else None | |
| ) | |
| return {"status": "ok", "mx_records": mx, "consistent": consistent, | |
| "a_fallback": False, "error": None} | |
| # No MX → A/AAAA fallback per RFC 5321 §5.1 | |
| try: | |
| async with httpx.AsyncClient(timeout=timeout) as client: | |
| a_task = _doh_query(client, GOOGLE_DOH, domain, "A") | |
| aaaa_task = _doh_query(client, GOOGLE_DOH, domain, "AAAA") | |
| a_res = await asyncio.gather(a_task, aaaa_task, return_exceptions=True) | |
| has_addr = any( | |
| isinstance(d, dict) and _parse_a_like(d, {1, 28}) for d in a_res | |
| ) | |
| except Exception: | |
| has_addr = False | |
| if has_addr: | |
| return {"status": "ok", "a_fallback": True, | |
| "mx_records": [{"host": domain, "priority": 0}], | |
| "consistent": None, "error": None} | |
| return {"status": "no_mail", "mx_records": [], "consistent": None, | |
| "a_fallback": False, "error": "Domain has no MX, A, or AAAA records"} | |
| # ---- domain-level TTL cache (shared-cache backed, memory tier always works) -- | |
| async def _cache_get(key: str) -> dict | None: | |
| try: | |
| from ..utils.cache import get_cache | |
| raw = await get_cache(key) | |
| if not raw: | |
| return None | |
| data = json.loads(raw) | |
| return data if isinstance(data, dict) else None | |
| except Exception: | |
| return None # cache outage → plain re-resolve | |
| async def _cache_set(key: str, payload: dict, ttl: int) -> None: | |
| try: | |
| from ..utils.cache import set_cache | |
| await set_cache(key, json.dumps(payload), ttl) | |
| except Exception: | |
| pass # a cache write failure must never break resolution | |
| def _mx_cache_key(domain: str) -> str: | |
| return f"doh:mx:{to_punycode(domain.strip().lower())}" | |
| async def resolve_mx_dual_cached(domain: str, timeout: Optional[float] = None) -> dict: | |
| """`resolve_mx_dual` with a domain-level TTL cache. | |
| Definitive answers cache for 24h; negative answers (NXDOMAIN / no mail) | |
| for 1h per the spec. Errors are never cached. | |
| """ | |
| domain = (domain or "").strip().lower() | |
| key = _mx_cache_key(domain) | |
| cached = await _cache_get(key) | |
| if cached is not None: | |
| cached = {**cached, "cache_hit": True} | |
| return cached | |
| result = await resolve_mx_dual(domain, timeout) | |
| if result.get("status") == "ok": | |
| await _cache_set(key, {**result, "cache_hit": False}, MX_TTL_SECONDS) | |
| elif result.get("status") in {"nxdomain", "no_mail"}: | |
| await _cache_set(key, {**result, "cache_hit": False}, NEGATIVE_TTL_SECONDS) | |
| return result | |