چون دیتاسنتر ها در جاهای حساس و فرودگاه ها هستن. این موضوع رو جدی بگیرید.

import argparse
import subprocess
import threading
import queue
import time
import signal

successful_servers = []
lock = threading.Lock()
stop_event = threading.Event()


def test_dns(server, timeout):
    try:
        start = time.perf_counter()

        result = subprocess.run(
            ["nslookup", "example.com", server],
            capture_output=True,
            text=True,
            timeout=timeout,
            shell=True  # مهم برای ویندوز
        )

        elapsed = time.perf_counter() - start
        output = (result.stdout + result.stderr).lower()

        if (
            result.returncode == 0
            and "timed out" not in output
            and "request to" not in output
            and "non-existent domain" not in output
        ):
            return elapsed

    except subprocess.TimeoutExpired:
        return None
    except Exception:
        return None

    return None


def worker(q, timeout):
    while not stop_event.is_set():
        try:
            server = q.get_nowait()
        except queue.Empty:
            return

        try:
            result = test_dns(server, timeout)
            if result is not None:
                with lock:
                    successful_servers.append((server, result))
                    print(f"[OK] {server} ({result:.3f}s)")
        finally:
            q.task_done()


def save_results(output_file):
    successful_servers.sort(key=lambda x: x[1])
    with open(output_file, "w", encoding="utf-8") as f:
        for server, _ in successful_servers:
            f.write(server + "\n")

    print(f"\nSaved {len(successful_servers)} servers to {output_file}")


def main():
    parser = argparse.ArgumentParser(description="Windows DNS Scanner")
    parser.add_argument("input_file")
    parser.add_argument("-o", "--output", default="dns-output.txt")
    parser.add_argument("-t", "--threads", type=int, default=20)
    parser.add_argument("--timeout", type=float, default=5)

    args = parser.parse_args()

    with open(args.input_file, "r") as f:
        servers = [line.strip() for line in f if line.strip()]

    q = queue.Queue()
    for s in servers:
        q.put(s)

    def signal_handler(sig, frame):
        print("\nStopping and saving results...")
        stop_event.set()

    signal.signal(signal.SIGINT, signal_handler)

    threads = []
    for _ in range(args.threads):
        t = threading.Thread(target=worker, args=(q, args.timeout), daemon=True)
        t.start()
        threads.append(t)

    q.join()
    save_results(args.output)


if __name__ == "__main__":
    main()

python dns-scanner.py nameservers.txt

python dns-scanner.py nameservers.txt -t 50 --timeout 3

nameservers.txt

8.8.8.8
1.1.1.1
223.5.5.5
9.9.9.9

cd Desktop

python fast-dns-scanner.py nameservers.txt

python fast-dns-scanner.py nameservers.txt -t 100 --timeout 1.5

import argparse
import socket
import threading
import queue
import time
import random
import struct

successful_servers = []
lock = threading.Lock()
stop_event = threading.Event()


def build_dns_query():
    transaction_id = random.randint(0, 65535)
    flags = 0x0100
    questions = 1
    header = struct.pack(">HHHHHH", transaction_id, flags, questions, 0, 0, 0)

    qname = b''.join(
        (bytes([len(part)]) + part.encode() for part in "example.com".split("."))
    ) + b"\x00"

    qtype = 1
    qclass = 1
    question = qname + struct.pack(">HH", qtype, qclass)

    return header + question


def test_dns(server, timeout):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.settimeout(timeout)

        query = build_dns_query()
        start = time.perf_counter()

        sock.sendto(query, (server, 53))
        data, _ = sock.recvfrom(512)

        elapsed = time.perf_counter() - start
        sock.close()

        if data:
            return elapsed

    except Exception:
        return None

    return None


def worker(q, timeout):
    while not stop_event.is_set():
        try:
            server = q.get_nowait()
        except queue.Empty:
            return

        try:
            result = test_dns(server, timeout)
            if result is not None:
                with lock:
                    successful_servers.append((server, result))
                    print(f"[OK] {server} ({result:.3f}s)")
        finally:
            q.task_done()


def save_results(output_file):
    successful_servers.sort(key=lambda x: x[1])
    with open(output_file, "w", encoding="utf-8") as f:
        for server, _ in successful_servers:
            f.write(server + "\n")

    print(f"\nSaved {len(successful_servers)} servers to {output_file}")


def main():
    parser = argparse.ArgumentParser(description="Fast UDP DNS Scanner (Windows)")
    parser.add_argument("input_file")
    parser.add_argument("-o", "--output", default="dns-output.txt")
    parser.add_argument("-t", "--threads", type=int, default=50)
    parser.add_argument("--timeout", type=float, default=2)

    args = parser.parse_args()

    with open(args.input_file, "r") as f:
        servers = [line.strip() for line in f if line.strip()]

    q = queue.Queue()
    for s in servers:
        q.put(s)

    threads = []
    for _ in range(args.threads):
        t = threading.Thread(target=worker, args=(q, args.timeout), daemon=True)
        t.start()
        threads.append(t)

    try:
        q.join()
    except KeyboardInterrupt:
        stop_event.set()

    save_results(args.output)


if __name__ == "__main__":
    main()