прочее Bulk Email MXrecord Lookup / Validation Script

[miamoder]

The Python script below performs bulk lookup of Mail Exchanger (MX) records for a list of email addresses. It makes use of Python's `asyncio` and `aiodns` libraries for asynchronous DNS queries, which enables it to process multiple requests concurrently, significantly improving performance. It also includes a progress bar, courtesy of the `tqdm` library, providing real-time feedback about the processing progress.

The email addresses are read from a file, with one address per line. The script then extracts the domain from each address and queries its MX records. If multiple MX records are found for a domain, all are captured. Any errors encountered during the query process, such as an absence of MX records or network issues, are also recorded.

The script supports multiple DNS servers to enhance reliability and potentially improve response times. If a query to one server fails, it will try the next server in the list. The results of the queries are written to an output file immediately after each lookup. This ensures that the data is saved even if the script gets interrupted for any reason.

*Use Cases:*
This script can be used for various purposes such as:
1. Email validation: By checking for the existence of an MX record, you can determine if a domain is set up to receive emails.
2. Server migration: If you're migrating email servers, you can use this script to monitor changes to MX records.
3. Cybersecurity: MX record lookups can be part of investigating email-related cybersecurity threats.
4. Research: This script can be used for research purposes to analyze the distribution of email server providers.

import asyncio
import aiodns
from tqdm import tqdm

email_file = 'emailleads.txt'
output_file = 'mx_records.txt'
dns_servers = ['8.8.8.8', '8.8.4.4', '1.1.1.1', '9.9.9.9']

resolvers = [aiodns.DNSResolver(nameservers=[server]) for server in dns_servers]

async def get_mx_record(email):
    domain = email.split('@')[-1]
    for resolver in resolvers:
        try:
            result = await resolver.query(domain, 'MX')
            return email, str(result[0].host)
        except Exception:
            pass
    return email, "Error: all DNS queries failed"

async def main():
    with open(email_file, 'r') as f:
        email_list = [line.strip() for line in f]
   
    tasks = [get_mx_record(email) for email in email_list]
   
    for i in tqdm(asyncio.as_completed(tasks), total=len(tasks), ncols=70):
        email, mx_record = await i
        with open(output_file, 'a') as out_f:
            out_f.write(f"{email}: {mx_record}\n")

loop = asyncio.get_event_loop()
loop.run_until_complete(main())