"""DNS Tester Blueprint — real DNS + ping via subprocess""" import asyncio import json import os import time import platform import socket import subprocess import re from dataclasses import dataclass, asdict from flask import Blueprint, jsonify, request, render_template, current_app bp = Blueprint("dns", __name__, url_prefix="/dns") # Cache file path _CACHE_DIR = os.path.dirname(os.path.abspath(__file__)) _PROJECT_ROOT = os.path.dirname(os.path.dirname(_CACHE_DIR)) # app/blueprints -> app -> project root RESULTS_PATH = os.path.join(_PROJECT_ROOT, "dns_results.json") DNS_CANDIDATES = [ ("Cloudflare", "1.1.1.1", "1.0.0.1"), ("Google Public DNS", "8.8.8.8", "8.8.4.4"), ("Quad9", "9.9.9.9", "149.112.112.112"), ("OpenDNS", "208.67.222.222", "208.67.220.220"), ("CleanBrowsing (Security)","185.228.168.9", "185.228.169.9"), ("AdGuard DNS", "94.140.14.14", "94.140.15.15"), ("NextDNS", "45.90.28.100","45.90.30.100"), ("Verisign", "64.6.64.6", "64.6.65.6"), ("Comodo Secure", "8.26.56.26", "8.20.247.20"), ("Level 3 / Lumen", "4.2.2.1", "4.2.2.2"), ("AT&T DNS", "68.94.156.156","68.94.157.157"), ("Mullvad", "194.242.2.2","194.242.2.3"), ("DNS.WATCH", "84.200.69.80","84.200.70.40"), ("Control D", "76.76.2.0", "76.76.10.0"), ("CZ.NIC", "89.46.222.1","89.46.222.2"), ("Norton ConnectSafe", "199.85.126.10","199.85.126.20"), ("Neustar UltraDNS", "156.154.70.1","156.154.71.1"), ("Hurricane Electric", "74.82.42.42","216.218.254.42"), ("Faranet+ (Gaming)", "193.110.81.0","193.110.81.1"), ("Freenom World", "80.80.80.80","80.80.81.80"), ] PING_COUNT = 4 PING_TIMEOUT = 3 @dataclass class DNSTestResult: name: str primary_ip: str secondary_ip: str dns_lookup_ms: float avg_ping_ms: float min_ping_ms: float max_ping_ms: float jitter_ms: float packet_loss_pct: float status: str def to_dict(self): return asdict(self) def run_ping_sync(ip: str, count: int = 1, timeout: int = 2): """Real system ping — synchronous (runs in executor).""" sys_name = platform.system() if sys_name == "Windows": cmd = ["ping", "-n", str(count), "-w", str(timeout * 1000), ip] else: cmd = ["ping", "-c", str(count), "-W", str(timeout), ip] try: proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 3) output = proc.stdout + proc.stderr if "100% packet loss" in output or "Request timed out" in output or "Unreachable" in output: return float("inf"), False matches = re.findall(r"time[=<]\s*([\d.]+)", output, re.IGNORECASE) if matches: return float(matches[0]), True m = re.search(r"rtt|min/avg/max/mdev\s*=\s*[\d.]+/([\d.]+)/[\d.]+/[\d.]+", output) if m: return float(m.group(1)), True return float("inf"), False except (subprocess.TimeoutExpired, Exception): return float("inf"), False async def measure_latency(server_ip: str): times = [] failures = 0 loop = asyncio.get_event_loop() for _ in range(PING_COUNT): t, ok = await loop.run_in_executor(None, run_ping_sync, server_ip, 1, PING_TIMEOUT) if ok: times.append(t) else: failures += 1 if not times: return {"avg": float("inf"), "min": float("inf"), "max": float("inf"), "jitter": float("inf"), "loss": 100.0, "samples": 0} avg = sum(times) / len(times) mn, mx = min(times), max(times) variance = sum((t - avg) ** 2 for t in times) / len(times) jitter = variance ** 0.5 return {"avg": avg, "min": mn, "max": mx, "jitter": jitter, "loss": (failures / PING_COUNT) * 100, "samples": len(times)} async def measure_dns_lookup(target_domain: str, server_ip: str, timeout: int = 5): """Async DNS lookup against specific resolver.""" old_timeout = socket.getdefaulttimeout() socket.setdefaulttimeout(timeout) try: t0 = __import__("time").perf_counter() info = await asyncio.get_event_loop().getaddrinfo( target_domain, 53, proto=socket.IPPROTO_UDP, flags=socket.AI_ADDRCONFIG, resolver=(server_ip, 53), ) elapsed = (time.perf_counter() - t0) * 1000 return elapsed, info[0][4][0] if info else None except Exception: return float("inf"), None finally: socket.setdefaulttimeout(old_timeout) async def benchmark_dns(server_ip: str): targets = ["api.riotgames.com", "connect.ubisoft.com", "battle.net", "steamcdn-a.akamaihd.net", "pubg-mobile.com"] latencies = [] for domain in targets: t, _ = await measure_dns_lookup(domain, server_ip) if t < float("inf"): latencies.append(t) if not latencies: return float("inf") return sum(latencies) / len(latencies) async def test_one(idx, total, name, primary, secondary): bar = f"[{idx}/{total}]" dns_res = await benchmark_dns(primary) lat = await measure_latency(primary) status = "ok" if lat["avg"] == float("inf") and dns_res == float("inf"): status = "fail" elif lat["avg"] == float("inf"): status = "timeout" elif lat["loss"] >= 50: status = "degraded" return DNSTestResult( name=name, primary_ip=primary, secondary_ip=secondary, dns_lookup_ms=round(dns_res, 1), avg_ping_ms=round(lat["avg"], 1), min_ping_ms=round(lat["min"], 1), max_ping_ms=round(lat["max"], 1), jitter_ms=round(lat["jitter"], 1), packet_loss_pct=round(lat["loss"], 1), status=status, ).to_dict() @bp.route("/") def index(): return render_template("dns_test.html") @bp.route("/test", methods=["GET", "POST"]) def run_test(): cache_path = RESULTS_PATH results = [] async def _run(): coros = [] for i, (name, primary, secondary) in enumerate(DNS_CANDIDATES, 1): coros.append(test_one(i, len(DNS_CANDIDATES), name, primary, secondary)) return await asyncio.gather(*coros, return_exceptions=True) raw = asyncio.run(_run()) for r in raw: if isinstance(r, dict) and "name" in r: results.append(r) results.sort(key=lambda x: x["min_ping_ms"]) payload = { "tested_at": time.strftime("%Y-%m-%d %H:%M:%S"), "count": len(results), "results": results, } try: with open(cache_path, "w", encoding="utf-8") as f: json.dump(payload, f, indent=2, ensure_ascii=False) except Exception: pass return jsonify(payload) @bp.route("/results") def get_results(): try: with open(RESULTS_PATH, "r", encoding="utf-8") as f: return jsonify(json.load(f)) except FileNotFoundError: return jsonify({"error": "no results yet", "results": []}), 404 except Exception: return jsonify({"error": "read error", "results": []}), 500 @bp.route("/config") def config(): """Return router-config-ready top DNS pair.""" try: with open(RESULTS_PATH, "r", encoding="utf-8") as f: data = json.load(f) except Exception: return jsonify({"error": "run /test first"}), 404 results = sorted(data["results"], key=lambda r: r["min_ping_ms"]) top = results[:3] if results else [] return jsonify({ "recommended": top[0] if top else None, "alternative": top[1] if len(top) > 1 else None, "all_sorted": top, "tested_at": data.get("tested_at", ""), })