import smtplib import ssl import imaplib import email import threading import time import random import os from concurrent.futures import ThreadPoolExecutor, as_completed from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from queue import Queue from colorama import init, Fore import socks import socket from tqdm import tqdm # do paska postępu init(autoreset=True) print(Fore.CYAN + "Ai by Revo - SMTP Checker + Proxy + IMAP Verification") # ----------------------- Konfiguracja ----------------------- MAX_WORKERS = 50 SEMAPHORE_LIMIT = 30 CONNECT_TIMEOUT = 8 SMTP_OP_TIMEOUT = 12 RETRY_COUNT = 1 BACKOFF_BASE = 1.5 PROXY_CHECK_TIMEOUT = 4 IMAP_POLL_INTERVAL = 5 # ----------------------- Funkcje ----------------------- def load_smtp_servers(filename): with open(filename, 'r', encoding='utf-8', errors='ignore') as f: return [line.strip().split('|') for line in f if line.strip() and len(line.strip().split('|')) == 4] def load_proxies(filename): with open(filename, 'r', encoding='utf-8', errors='ignore') as f: return [line.strip() for line in f if line.strip()] def set_proxy(proxy_type, proxy_str): ip, port = proxy_str.split(':') port = int(port) types = { 'http': socks.HTTP, 'https': socks.HTTP, 'socks4': socks.SOCKS4, 'socks5': socks.SOCKS5 } socks.setdefaultproxy(types[proxy_type], ip, port) socket.socket = socks.socksocket def reset_socket(): import _socket socket.socket = _socket.socket def test_proxy(proxy_type, proxy_str): """Szybkie testowanie proxy — teraz równolegle""" ip, port = proxy_str.split(':') port = int(port) try: set_proxy(proxy_type, proxy_str) # Gmail IPv4 - unikamy IPv6 problemów PySocks sock = socket.create_connection(("142.250.195.108", 587), timeout=PROXY_CHECK_TIMEOUT) sock.close() reset_socket() return True except: reset_socket() return False def send_email(smtp_info, to_email, subject, content, use_proxy, proxy_type, proxy_list, success_queue, checked_counter, semaphore): host, port, user, password = smtp_info port = int(port) message = MIMEMultipart() message['From'] = user message['To'] = to_email message['Subject'] = subject message.attach(MIMEText(content, 'plain')) for attempt in range(RETRY_COUNT + 1): try: if use_proxy: proxy = random.choice(proxy_list) set_proxy(proxy_type, proxy) with semaphore: context = ssl.create_default_context() if port == 465: server = smtplib.SMTP_SSL(host, port, timeout=SMTP_OP_TIMEOUT, context=context) else: server = smtplib.SMTP(host, port, timeout=SMTP_OP_TIMEOUT) server.starttls(context=context) server.login(user, password) server.sendmail(user, to_email, message.as_string()) server.quit() success_queue.put('|'.join(smtp_info)) print(Fore.GREEN + f"[+] Email sent: {host}|{port}|{user}|{password}") break except Exception as e: wait = BACKOFF_BASE * (attempt + 1) print(Fore.YELLOW + f"[WARN] Attempt {attempt+1} failed for {user} -> {e}. Backing off {wait:.1f}s") time.sleep(wait) finally: if use_proxy: reset_socket() checked_counter.append(1) def imap_check(server, user, password, subject, duration_minutes, success_queue): found_senders = set() duration_seconds = duration_minutes * 60 start_time = time.time() while True: # przerwanie po czasie if time.time() - start_time >= duration_seconds: print(Fore.GREEN + "[+] Zakończono monitorowanie IMAP (czas minął).") break try: print(Fore.CYAN + "[i] Łączenie z IMAP...") mail = imaplib.IMAP4_SSL(server) mail.login(user, password) print(Fore.GREEN + "[+] Zalogowano do skrzynki IMAP.") mail.select("inbox") # pętla w ramach aktywnego połączenia while time.time() - start_time < duration_seconds: try: result, data = mail.search(None, f'(SUBJECT "{subject}")') if result == 'OK': ids = data[0].split() for msg_id in ids: result, msg_data = mail.fetch(msg_id, '(RFC822)') if result == 'OK': msg = email.message_from_bytes(msg_data[0][1]) from_addr = msg['From'] if from_addr not in found_senders: print(Fore.YELLOW + f"[+] Email doręczony od: {from_addr}") found_senders.add(from_addr) success_queue.put(from_addr) time.sleep(IMAP_POLL_INTERVAL) except Exception as e: print(Fore.RED + f"[!] Błąd IMAP w aktywnej sesji: {e}") print(Fore.CYAN + "[i] Próba ponownego połączenia...") break # wyjście z sesji → reconnect try: mail.logout() print(Fore.GREEN + "[+] Wylogowano z IMAP.") except: pass except Exception as e: print(Fore.RED + f"[-] IMAP login failed: {e}") print(Fore.CYAN + "[i] Ponawiam próbę za 5 sekund...") time.sleep(5) continue # ----------------------- Main ----------------------- def main(): smtp_file = input("Enter SMTP servers filename: ") threads = int(input("Enter number of threads: ")) to_email = input("Enter recipient email: ") subject = input("Enter email subject: ") content = input("Enter email content: ") use_proxy = input("Use proxy? (y/n): ").lower() == 'y' proxy_list = [] proxy_type = '' if use_proxy: proxy_file = input("Enter proxy list filename: ") proxy_type = input("Proxy type (http/https/socks4/socks5): ").lower() proxy_list_raw = load_proxies(proxy_file) print(Fore.CYAN + f"[i] Testowanie proxy równolegle...") # ------------ NOWA SZYBKA WERSJA ------------- good = [] with ThreadPoolExecutor(max_workers=50) as ex: futures = {ex.submit(test_proxy, proxy_type, p): p for p in proxy_list_raw} for future in tqdm(as_completed(futures), total=len(futures), desc="Proxy test"): p = futures[future] try: if future.result(): good.append(p) except: pass proxy_list = good print(Fore.CYAN + f"[i] Działające proxy: {len(proxy_list)}/{len(proxy_list_raw)}") if not proxy_list: print(Fore.RED + "Brak działających proxy. Kończę.") return smtp_servers = load_smtp_servers(smtp_file) success_queue = Queue() checked_counter = [] semaphore = threading.Semaphore(SEMAPHORE_LIMIT) # Wysyłanie SMTP with ThreadPoolExecutor(max_workers=threads) as executor: futures = [] for smtp in smtp_servers: if len(smtp) != 4: continue futures.append( executor.submit(send_email, smtp, to_email, subject, content, use_proxy, proxy_type, proxy_list, success_queue, checked_counter, semaphore)) # Pasek postępu sprawdzonych SMTP for _ in tqdm(as_completed(futures), total=len(futures), desc="Sprawdzone SMTP"): pass print(Fore.CYAN + "[i] Wysyłka zakończona. Rozpoczynam IMAP verification.") imap_server = input("IMAP server: ") imap_user = input("IMAP user: ") imap_pass = input("IMAP password: ") duration = int(input("Minutes to check inbox for replies: ")) imap_check(imap_server, imap_user, imap_pass, subject, duration, success_queue) # Zapis wyników good_file = 'good.txt' good_set = set() while not success_queue.empty(): good_set.add(success_queue.get()) with open(good_file, 'w', encoding='utf-8') as f: for smtp in good_set: f.write(smtp + '\n') print(Fore.GREEN + f"[+] Zapisano {len(good_set)} działających SMTP do good.txt") print(Fore.CYAN + "Ai by Revo - koniec programu") if __name__ == "__main__": main()