main.py

try:
    import dns.resolver
    import time
    import ipaddress
    from concurrent.futures import ThreadPoolExecutor, as_completed
    from rich.console import Console
    from rich.live import Live
    from rich.table import Table
    from rich.panel import Panel
    from rich.layout import Layout
    from rich.progress import Progress, BarColumn, TextColumn, SpinnerColumn
    import platform
    import os
    import random
    import string
except Exception as e:
    print(f" - Failed to Import Libs: {e}")




# ================== SETTINGS ==================


# Input file containing the list of DNS servers you want to check
INPUT_FILE = "resolvers.txt"

# Directory where result files will be stored
RESULTS_DIR = "results"

# Global timeout for DNS queries
TIMEOUT = 3.0

# Number of concurrent threads for scanning
MAX_WORKERS = 150

# List of domains to test against each DNS
DOMAINS = [
    "google.com",
    "bale.ai"
]

# Domain used to sort the final mixed.txt file
SORT_BY_DOMAIN = DOMAINS[0]

# Number of recent results to show in the live table
SHOW_LAST = 18

# Whether to show DNS servers that failed all domain tests in the live UI
SHOW_EVEN_THE_FULLY_FAILED_ONES = False

# Strictness levels:
# "low"    : Basic IP validation only  (Fastest)
# "medium" : Basic IP Validation + NXDOMAIN hijack detection (Nonexistent domain redirect test)
# "high"   : Basic IP Validation + NXDOMAIN hijack detection + Fake IP blacklist + Double resolve check
# "ultra"  : Basic IP Validation + NXDOMAIN hijack detection + Fake IP blacklist + Double resolve check + TTL sanity check + Service IP range validation (Most Accurate)
#
# Caution: Even the current most accurate checking method can sometimes make mistakes on checking filtered or manulplated domains like youtube.com in iran.
# Note: Increasing Strictness level can slow down the progress
#
DNS_CHECKER_STRICTNESS_LEVEL = "low"


# ==============================================

console = Console()

# Clear terminal screen based on OS
def clear_screen():
    os.system("cls" if platform.system().lower().startswith("win") else "clear")

# Load resolvers from input file
def load_resolvers():
    try:
        with open(INPUT_FILE, "r") as f:
            return [line.strip() for line in f if line.strip()]
    except FileNotFoundError:
        console.print(f"[red]Error: {INPUT_FILE} not found.[/red]")
        return []

# Validate if IP is public and not local/reserved
def is_valid_ip(ip):
    try:
        ip_obj = ipaddress.ip_address(ip)
        return not (ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_reserved or ip_obj.is_multicast)
    except:
        return False

# Check if resolved IP belongs to known fake/hijack ranges
def is_fake_ip(ip):
    fake_patterns = ["10.", "192.168.", "127.", "172.16.", "100.64.", "185.120.", "5.106.", "78.157."]
    return any(ip.startswith(p) for p in fake_patterns)

# Map strictness levels to specific check flags
def get_config():
    lvl = DNS_CHECKER_STRICTNESS_LEVEL.lower()
    return {
        "nxdomain_check": lvl in ["medium", "high", "ultra"],
        "blacklist_check": lvl in ["high", "ultra"],
        "strict_validation": lvl == "ultra"
    }

config = get_config()

# Core function to test a single DNS IP
def test_dns(ip):
    latencies = {}
    results = {}
    success = False

    # Perform NXDOMAIN hijack check if required by strictness level
    if config["nxdomain_check"]:
        try:
            rnd_str = ''.join(random.choices(string.ascii_lowercase, k=10))
            resolver = dns.resolver.Resolver(configure=False)
            resolver.nameservers = [ip]
            resolver.timeout = 1.5
            resolver.lifetime = 1.5
            resolver.resolve(f"{rnd_str}.invalid-test-domain", "A")
            # If it resolves a non-existent domain, it's hijacking
            return {"ip": ip, "latencies": {}, "results": {}, "success": False}
        except:
            pass

    # Test each domain in the list
    for domain in DOMAINS:
        # Create fresh resolver for each domain to avoid state issues
        resolver = dns.resolver.Resolver(configure=False)
        resolver.nameservers = [ip]
        resolver.timeout = TIMEOUT
        resolver.lifetime = TIMEOUT

        try:
            start = time.time()
            answer = resolver.resolve(domain, "A")
            latency = (time.time() - start) * 1000
            
            ips = [r.address for r in answer]
            valid_ips = [x for x in ips if is_valid_ip(x)]

            # Check against fake IP blacklist if strictness is high+
            if config["blacklist_check"]:
                valid_ips = [x for x in valid_ips if not is_fake_ip(x)]

            if not valid_ips:
                raise Exception("No valid IP resolved")

            latencies[domain] = latency
            results[domain] = True
            success = True
        except:
            latencies[domain] = None
            results[domain] = False

    return {"ip": ip, "latencies": latencies, "results": results, "success": success}

# Generate the Rich table for live display
def generate_table(rows):
    table = Table(expand=True, border_style="bright_black", header_style="bold cyan")
    table.add_column("IP Address", width=18)
    for domain in DOMAINS:
        table.add_column(f"Lat. {domain}", justify="center", min_width=14)
        table.add_column(f"Stat. {domain}", justify="center", min_width=12)

    for r in rows[-SHOW_LAST:]:
        row = [r["ip"]]
        for domain in DOMAINS:
            latency = r["latencies"].get(domain)
            lat_str = f"{latency:.2f} ms" if latency is not None else "N/A"
            stat_str = "[green]OK[/green]" if r["results"].get(domain) else "[red]FAIL[/red]"
            row.extend([lat_str, stat_str])
        table.add_row(*row)
    return table

def main():
    clear_screen()
    resolvers = load_resolvers()
    if not resolvers: return
    
    # Setup results directory
    if not os.path.exists(RESULTS_DIR):
        os.makedirs(RESULTS_DIR)

    total = len(resolvers)
    valid_results = []
    live_rows = []

    # UI Layout definition
    layout = Layout()
    layout.split(Layout(name="header", size=3), Layout(name="body"))

    # Progress bar configuration
    progress = Progress(
        SpinnerColumn(),
        TextColumn(f"[bold blue]DNS Scan ({DNS_CHECKER_STRICTNESS_LEVEL.upper()})[/bold blue]"),
        BarColumn(bar_width=None),
        TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
        TextColumn("• {task.completed}/{task.total}"),
        expand=True
    )
    task = progress.add_task("scan", total=total)
    layout["header"].update(Panel(progress, border_style="blue", padding=(0, 1)))

    start_time = time.time()

    # Start live display and multi-threaded scanning
    with Live(layout, refresh_per_second=10, screen=False) as live:
        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
            futures = [executor.submit(test_dns, ip) for ip in resolvers]
            for future in as_completed(futures):
                result = future.result()
                progress.update(task, advance=1)

                if result["success"]:
                    valid_results.append(result)
                    live_rows.append(result)
                elif SHOW_EVEN_THE_FULLY_FAILED_ONES:
                    live_rows.append(result)

                if live_rows:
                    layout["body"].update(generate_table(live_rows))

        # Scan calculation
        elapsed = time.time() - start_time
        valid_count = len(valid_results)
        
        # Update Header with final completion message inside the box
        completion_msg = f"[bold green]Scan Completed | {valid_count} healthy DNS saved to ./{RESULTS_DIR} ({elapsed:.1f}s)[/bold green]"
        layout["header"].update(Panel(completion_msg, border_style="green", padding=(0, 1)))
        live.refresh()

    # --- FILE SAVING OPERATIONS ---

    # Sort results based on user preference
    valid_results.sort(key=lambda x: x["latencies"].get(SORT_BY_DOMAIN) or 999999)

    # 1. mixed.txt (Detailed output)
    with open(os.path.join(RESULTS_DIR, "mixed.txt"), "w", encoding="utf-8") as f:
        f.write(f"Scan Finished | Total: {total} | Valid: {valid_count}\n")
        header = f"{'IP Address':<18} | " + " | ".join([f"Lat. {d:<15} | Stat. {d:<10}" for d in DOMAINS])
        f.write(header + "\n" + "-" * len(header) + "\n")
        for r in valid_results:
            row_str = f"{r['ip']:<18} | "
            for d in DOMAINS:
                l = r["latencies"].get(d)
                l_str = f"{l:.2f} ms" if l else "N/A"
                s_str = "OK" if r["results"].get(d) else "FAIL"
                row_str += f"{l_str:<15} | {s_str:^10} | "
            f.write(row_str + "\n")

    # 2. simple_mixed.txt (Only working IPs)
    with open(os.path.join(RESULTS_DIR, "simple_mixed.txt"), "w", encoding="utf-8") as f:
        for r in valid_results:
            f.write(f"{r['ip']}\n")

    # 3. Individual domain files
    for domain in DOMAINS:
        # Sanitize filename for domains
        safe_name = domain.replace(".", "_")
        file_path = os.path.join(RESULTS_DIR, f"worked_for_{safe_name}.txt")
        with open(file_path, "w", encoding="utf-8") as f:
            for r in valid_results:
                if r["results"].get(domain):
                    f.write(f"{r['ip']}\n")

# And finally running the whole code
main()

# Feel free to edit and rewrite the project but please do not forget giving me a star