import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
import requests
import json
import time
import threading # To run network requests in the background
# --- API Configuration (will be managed by GUI) ---
MAILSTORM_API_URL = "http://ypwri6hshlwgxltrzorrcptslirxk7qkrvrzdi6i5fvrm36kkgw764yd.onion"
TOR_PROXY = {
'http': 'socks5h://127.0.0.1:9050',
'https': 'socks5h://127.0.0.1:9050'
}
# --- API Interaction Functions (modified to return results/errors) ---
def check_health_api(access_key):
"""
Checks the health of the Mailstorm API.
Returns a tuple: (success_boolean, message_string)
"""
endpoint = f"{MAILSTORM_API_URL}/health"
try:
response = requests.get(endpoint, proxies=TOR_PROXY, timeout=10)
response.raise_for_status()
if response.text.strip().lower() == "true":
return True, "Service is healthy."
else:
return False, f"Service is not healthy. Response: {response.text}"
except requests.exceptions.Timeout:
return False, "Health Check Error: Request timed out."
except requests.exceptions.RequestException as e:
return False, f"Health Check Error: {e}"
def check_access_usage_api(access_key):
"""
Checks the usage statistics for the provided access key.
Returns a tuple: (success_boolean, message_string)
"""
endpoint = f"{MAILSTORM_API_URL}/access_usage"
payload = {"access": access_key}
try:
response = requests.post(endpoint, proxies=TOR_PROXY, json=payload, timeout=10)
response.raise_for_status()
return True, f"Access Usage: {response.text}"
except requests.exceptions.Timeout:
return False, "Access Usage Error: Request timed out."
except requests.exceptions.RequestException as e:
return False, f"Access Usage Error: {e}"
def bomb_mail_api(first_name: str, last_name: str, email: str, access_key: str, max_retries=5, retry_delay=5):
"""
Subscribes an email address to mailing lists.
Retries on specific errors that should be ignored.
Returns a tuple: (success_boolean, message_string)
"""
endpoint = f"{MAILSTORM_API_URL}/bombmail"
payload = {
"first_name": first_name,
"last_name": last_name,
"email": email,
"access": access_key
}
ignored_errors = ["mail not sent", "csrf error", "link non existent"]
for attempt in range(max_retries + 1):
try:
response = requests.post(endpoint, proxies=TOR_PROXY, json=payload, timeout=20)
# Successful response
if response.status_code == 200 and response.text.strip() == "sent":
return True, f"Bomb Mail Success for {email}: Sent."
# Handle specific ignored errors
response_text = response.text.strip().lower()
if any(err in response_text for err in ignored_errors):
message = f"Bomb Mail Warning for {email}: Ignored error '{response_text}'. Retrying ({attempt+1}/{max_retries})..."
if attempt < max_retries:
time.sleep(retry_delay)
continue # Retry the request
else:
return False, f"Bomb Mail Failed for {email} after {max_retries} retries due to ignored errors."
# Handle other bad requests
elif response.status_code == 400:
error_detail = response.json() if response.headers.get('Content-Type') == 'application/json' else response.text
return False, f"Bomb Mail Error for {email}: Bad Request. Response: {error_detail}"
# Handle server errors
elif response.status_code == 500:
error_detail = response.json() if response.headers.get('Content-Type') == 'application/json' else response.text
return False, f"Bomb Mail Error for {email}: Internal Server Error. Response: {error_detail}"
else:
response.raise_for_status() # Raise for any other unexpected status codes
except requests.exceptions.Timeout:
message = f"Bomb Mail Error for {email}: Request timed out. Retrying ({attempt+1}/{max_retries})..."
if attempt < max_retries:
time.sleep(retry_delay)
continue
else:
return False, f"Bomb Mail Failed for {email} after {max_retries} retries due to timeout."
except requests.exceptions.RequestException as e:
message = f"Bomb Mail Error for {email}: {e}. Retrying ({attempt+1}/{max_retries})..."
if attempt < max_retries:
time.sleep(retry_delay)
continue
else:
return False, f"Bomb Mail Failed for {email} after {max_retries} retries."
return False, "Bomb Mail Failed: Unexpected error." # Fallback
# --- GUI Application ---
class MailstormGUI:
def __init__(self, master):
self.master = master
master.title("Mailstorm API Client")
master.geometry("600x550") # Adjusted size
self.log_messages = []
# --- Access Key Input ---
ttk.Label(master, text="Access Key:").grid(row=0, column=0, padx=5, pady=5, sticky="w")
self.access_key_entry = ttk.Entry(master, width=50)
self.access_key_entry.grid(row=0, column=1, padx=5, pady=5, columnspan=2, sticky="ew")
self.access_key_entry.insert(0, "YOUR_ACCESS_KEY_HERE") # Placeholder
# --- Input Fields for Bomb Mail ---
ttk.Label(master, text="First Name:").grid(row=1, column=0, padx=5, pady=5, sticky="w")
self.first_name_entry = ttk.Entry(master, width=40)
self.first_name_entry.grid(row=1, column=1, padx=5, pady=5, columnspan=2, sticky="ew")
self.first_name_entry.insert(0, "John") # Default value
ttk.Label(master, text="Last Name:").grid(row=2, column=0, padx=5, pady=5, sticky="w")
self.last_name_entry = ttk.Entry(master, width=40)
self.last_name_entry.grid(row=2, column=1, padx=5, pady=5, columnspan=2, sticky="ew")
self.last_name_entry.insert(0, "Doe") # Default value
ttk.Label(master, text="Email:").grid(row=3, column=0, padx=5, pady=5, sticky="w")
self.email_entry = ttk.Entry(master, width=40)
self.email_entry.grid(row=3, column=1, padx=5, pady=5, columnspan=2, sticky="ew")
self.email_entry.insert(0, "john.doe@example.com") # Default value
# --- Buttons ---
self.check_health_button = ttk.Button(master, text="Check Health", command=self.run_check_health)
self.check_health_button.grid(row=4, column=0, padx=5, pady=10, sticky="ew")
self.check_usage_button = ttk.Button(master, text="Check Usage", command=self.run_check_usage)
self.check_usage_button.grid(row=4, column=1, padx=5, pady=10, sticky="ew")
self.send_bombmail_button = ttk.Button(master, text="Send Bomb Mail", command=self.run_bomb_mail)
self.send_bombmail_button.grid(row=4, column=2, padx=5, pady=10, sticky="ew")
# --- Output Log Area ---
ttk.Label(master, text="API Log:").grid(row=5, column=0, padx=5, pady=5, sticky="w")
self.log_area = scrolledtext.ScrolledText(master, wrap=tk.WORD, width=70, height=15, state='disabled')
self.log_area.grid(row=6, column=0, columnspan=3, padx=5, pady=5, sticky="nsew")
# Configure grid weights for resizing
master.grid_columnconfigure(1, weight=1)
master.grid_columnconfigure(2, weight=1)
master.grid_rowconfigure(6, weight=1)
self.update_log("GUI initialized. Ensure Tor is running and proxy is accessible (127.0.0.1:9050).")
self.update_log("Please enter your Access Key.")
def update_log(self, message):
"""Appends a message to the log area."""
self.log_messages.append(message)
self.log_area.config(state='normal')
self.log_area.insert(tk.END, message + "\n")
self.log_area.see(tk.END) # Auto-scroll to the bottom
self.log_area.config(state='disabled')
self.master.update_idletasks() # Force GUI update
def run_api_action(self, action_func, *args):
"""Helper to run API functions in a separate thread and update GUI."""
self.update_log(f"Starting action: {action_func.__name__}...")
# Disable buttons during API call
self.check_health_button.config(state=tk.DISABLED)
self.check_usage_button.config(state=tk.DISABLED)
self.send_bombmail_button.config(state=tk.DISABLED)
def task():
try:
result, message = action_func(*args)
self.update_log(message)
if not result and "Error" in message:
messagebox.showerror("API Error", message)
except Exception as e:
self.update_log(f"An unexpected error occurred in thread: {e}")
messagebox.showerror("Application Error", f"An unexpected error occurred: {e}")
finally:
# Re-enable buttons after API call
self.check_health_button.config(state=tk.NORMAL)
self.check_usage_button.config(state=tk.NORMAL)
self.send_bombmail_button.config(state=tk.NORMAL)
thread = threading.Thread(target=task)
thread.daemon = True # Allows the program to exit even if thread is running
thread.start()
def run_check_health(self):
access_key = self.access_key_entry.get()
if not access_key or access_key == "YOUR_ACCESS_KEY_HERE":
messagebox.showwarning("Input Missing", "Please enter your Access Key.")
return
self.run_api_action(check_health_api, access_key)
def run_check_usage(self):
access_key = self.access_key_entry.get()
if not access_key or access_key == "YOUR_ACCESS_KEY_HERE":
messagebox.showwarning("Input Missing", "Please enter your Access Key.")
return
self.run_api_action(check_access_usage_api, access_key)
def run_bomb_mail(self):
access_key = self.access_key_entry.get()
first_name = self.first_name_entry.get()
last_name = self.last_name_entry.get()
email = self.email_entry.get()
if not access_key or access_key == "YOUR_ACCESS_KEY_HERE":
messagebox.showwarning("Input Missing", "Please enter your Access Key.")
return
if not first_name or not last_name or not email:
messagebox.showwarning("Input Missing", "Please fill in all recipient details (First Name, Last Name, Email).")
return
# Basic email format validation
if "@" not in email or "." not in email.split('@')[1]:
messagebox.showwarning("Invalid Input", "Please enter a valid email address.")
return
self.run_api_action(bomb_mail_api, first_name, last_name, email, access_key)
# --- Main Execution ---
if __name__ == "__main__":
root = tk.Tk()
gui = MailstormGUI(root)
root.mainloop()
