بروكسي للتسويق بالبريد الإلكتروني مع منافذ SMTP مفتوحة

#!/usr/bin/env python3
"""SMTP reachability checker for Advanced.name proxies.

A legitimate mail setup needs to know which proxies can actually reach an SMTP
server. This opens a SOCKS5 tunnel to the mail host on ports 25, 465 and 587
through each proxy and reads the greeting, so you keep only the addresses that
work. It does not send mail. Standard library only. Run: python3 smtp_reach.py
"""

# ================================ CONFIG ====================================
PROXY_LIST = "proxies.txt"   # one ip:port per line, or a CSV export (';' separated)
SMTP_HOST = "smtp.gmail.com"  # the mail server you plan to relay through
SMTP_PORTS = (587, 465, 25)   # ports to test on that host
THREADS = 100
TIMEOUT = 12
OUT_FILE = "smtp_ok.csv"     # proxy;port;banner is written here
# ============================================================================

import csv
import socket
import sys
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed


def load_list(path):
    out = []
    with open(path, encoding="utf-8", errors="replace") as fh:
        for line in fh:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            if ";" in line:                       # CSV row: ip;port;...
                parts = line.split(";")
                if len(parts) >= 2 and parts[1].isdigit():
                    out.append(parts[0] + ":" + parts[1])
            elif ":" in line:
                out.append(line)
    return out


def socks5_connect(proxy, host, port):
    """Open host:port through a no-auth SOCKS5 proxy, return the connected socket."""
    ip, pport = proxy.rsplit(":", 1)
    sock = socket.create_connection((ip, int(pport)), TIMEOUT)
    sock.settimeout(TIMEOUT)
    sock.sendall(b"\x05\x01\x00")
    if sock.recv(2) != b"\x05\x00":
        raise ConnectionError("socks5 handshake")
    host_bytes = host.encode()
    sock.sendall(b"\x05\x01\x00\x03" + bytes([len(host_bytes)]) + host_bytes + port.to_bytes(2, "big"))
    reply = sock.recv(4)
    if len(reply) < 2 or reply[1] != 0:
        raise ConnectionError("socks5 connect rejected")
    sock.recv(256)                                # drain the bound-address tail
    return sock


def reachable(proxy):
    """Return (port, banner) for the first SMTP port that answers, or (0, reason)."""
    for port in SMTP_PORTS:
        try:
            sock = socks5_connect(proxy, SMTP_HOST, port)
        except OSError:
            continue
        try:
            if port == 465:                       # implicit TLS: no plaintext banner
                sock.close()
                return port, "connected (tls)"
            banner = sock.recv(256).decode("latin-1", "replace").strip()
            sock.close()
            if banner.startswith("220"):
                return port, banner[:80]
        except OSError:
            pass
    return 0, "no smtp port reachable"


def main():
    proxies = load_list(PROXY_LIST)
    if not proxies:
        sys.exit("List is empty.")
    print("host %s, ports %s, proxies %d" % (SMTP_HOST, SMTP_PORTS, len(proxies)))
    ok, lock = [], threading.Lock()

    def work(proxy):
        port, info = reachable(proxy)
        if port:
            with lock:
                ok.append((proxy, port, info))
        return proxy, port, info

    with ThreadPoolExecutor(max_workers=THREADS) as pool:
        for fut in as_completed([pool.submit(work, p) for p in proxies]):
            proxy, port, info = fut.result()
            print("%-22s %s" % (proxy, ("%d %s" % (port, info)) if port else info))

    with open(OUT_FILE, "w", encoding="utf-8", newline="") as out:
        writer = csv.writer(out, delimiter=";")
        writer.writerow(["proxy", "port", "banner"])
        writer.writerows(ok)
    print("reachable: %d of %d -> %s" % (len(ok), len(proxies), OUT_FILE))


if __name__ == "__main__":
    main()