#!/usr/bin/env python3 """ PyFlood Suite v1.2.5 Educational Network Testing Tool Command Line Interface Version """ import threading import time import json import random import os from datetime import datetime from typing import List, Dict, Optional from dataclasses import dataclass, asdict from enum import Enum import argparse # Color codes for terminal output class Colors: HEADER = '\033[95m' BLUE = '\033[94m' CYAN = '\033[96m' GREEN = '\033[92m' YELLOW = '\033[93m' RED = '\033[91m' MAGENTA = '\033[35m' BOLD = '\033[1m' UNDERLINE = '\033[4m' END = '\033[0m' class AttackType(Enum): STEAM_FRIEND = "steam_friend" STEAM_MESSAGE = "steam_message" STEAM_COMMENT = "steam_comment" IP_UDP = "ip_udp" IP_SYN = "ip_syn" IP_HTTP = "ip_http" IP_ICMP = "ip_icmp" EMAIL_FLOOD = "email_flood" @dataclass class SteamConfig: targets: List[str] attack_type: str count: int message: str @dataclass class IPConfig: targets: List[str] attack_type: str duration: int port: int = 80 @dataclass class EmailConfig: targets: List[str] emails_per_target: int delay: int subject: str body: str class ConsoleLogger: def __init__(self): self.log_history = [] def log(self, message: str, color: str = Colors.END, prefix: str = ">>"): timestamp = datetime.now().strftime("%H:%M:%S") formatted = f"{Colors.CYAN}[{timestamp}]{Colors.END} {color}{prefix} {message}{Colors.END}" print(formatted) self.log_history.append({ "time": timestamp, "message": message, "type": color }) def success(self, message: str): self.log(message, Colors.GREEN) def error(self, message: str): self.log(message, Colors.RED) def warning(self, message: str): self.log(message, Colors.YELLOW) def info(self, message: str): self.log(message, Colors.BLUE) def debug(self, message: str): self.log(message, Colors.MAGENTA) class AttackSimulator: def __init__(self, logger: ConsoleLogger): self.logger = logger self.active_attacks = {} self.attack_counter = 0 self.lock = threading.Lock() def steam_flood(self, config: SteamConfig, attack_id: str): """Simulate Steam ID flooding""" self.logger.info(f"Initializing Steam ID flood attack #{attack_id}...") self.logger.info(f"Targets: {len(config.targets)} Steam IDs") self.logger.info(f"Attack type: {config.attack_type}") time.sleep(1) if config.attack_type == "Friend Requests": self.logger.info("Connecting to Steam API proxy...") time.sleep(0.8) self.logger.success("Connected to Steam relay servers") for i in range(config.count): if attack_id not in self.active_attacks: self.logger.warning("Attack stopped by user") break target = random.choice(config.targets) with self.lock: self.attack_counter += 1 if config.attack_type == "Friend Requests": actions = [ f"Sent friend request to {target}", f"Sending add friend packet to Steam ID {target}", f"Friend request queued for {target}" ] elif config.attack_type == "Message Flood": actions = [ f"Sent message to {target}: '{config.message}'", f"Message delivered to {target}", f"Chat packet sent to {target}" ] else: # Profile Comments actions = [ f"Posted comment on {target}'s profile", f"Profile comment submitted to {target}", f"Comment flood sent to {target}" ] self.logger.success(random.choice(actions)) time.sleep(random.uniform(0.5, 1.5)) self.logger.info(f"Steam attack #{attack_id} completed. Sent {config.count} requests.") def ip_flood(self, config: IPConfig, attack_id: str): """Simulate IP flooding""" self.logger.info(f"Initializing IP flood attack #{attack_id}...") self.logger.warning(f"Target IPs: {', '.join(config.targets)}") self.logger.info(f"Attack vector: {config.attack_type}") time.sleep(1) if config.attack_type == "UDP Flood": self.logger.info("Preparing UDP payload...") elif config.attack_type == "SYN Flood": self.logger.info("Initializing SYN packet generator...") elif config.attack_type == "HTTP Flood": self.logger.info("Preparing HTTP request headers...") elif config.attack_type == "ICMP Flood": self.logger.info("Loading ICMP packet structure...") self.logger.success("Attack modules loaded successfully") time.sleep(0.5) end_time = time.time() + config.duration packet_count = 0 while time.time() < end_time: if attack_id not in self.active_attacks: self.logger.warning("Attack interrupted by user") break target = random.choice(config.targets) if config.attack_type == "UDP Flood": size = random.randint(64, 1024) msg = f"UDP packet sent to {target}:{config.port} ({size} bytes)" elif config.attack_type == "SYN Flood": msg = f"SYN request sent to {target}:{config.port} [SYN_SENT]" elif config.attack_type == "HTTP Flood": path = random.choice(["/", "/index.html", "/api", "/login", "/admin"]) msg = f"HTTP GET {path} sent to {target}:{config.port} [200 OK]" else: # ICMP msg = f"ICMP echo request to {target} [ttl=64]" self.logger.success(msg) packet_count += 1 # Fast transmission for IP floods time.sleep(random.uniform(0.01, 0.1)) self.logger.info(f"IP flood completed. Sent {packet_count} packets over {config.duration} seconds.") def email_flood(self, config: EmailConfig, attack_id: str): """Simulate Email flooding""" self.logger.info(f"Initializing Email flood attack #{attack_id}...") self.logger.info(f"Target addresses: {len(config.targets)}") self.logger.info(f"Emails per target: {config.emails_per_target}") time.sleep(1) self.logger.info("Connecting to SMTP relay...") time.sleep(0.8) self.logger.success("SMTP connection established") self.logger.info(f"Subject: {config.subject}") sent_count = 0 for target in config.targets: for i in range(config.emails_per_target): if attack_id not in self.active_attacks: self.logger.warning("Attack stopped by user") return with self.lock: sent_count += 1 self.logger.success(f"Email #{i+1} sent to {target}") if config.delay > 0: time.sleep(config.delay / 1000) self.logger.info(f"Email flood completed. Sent {sent_count} messages.") class PyFloodSuite: def __init__(self): self.logger = ConsoleLogger() self.simulator = AttackSimulator(self.logger) self.running = True self.config_file = "pyflood_config.json" def clear_screen(self): os.system('cls' if os.name == 'nt' else 'clear') def print_banner(self): banner = f""" {Colors.BLUE}{Colors.BOLD} ╔═══════════════════════════════════════════════════════════════╗ ║ ║ ║ ██████╗ ██╗ ██╗███████╗██╗ ██████╗ ██████╗ ██████╗ ║ ║ ██╔══██╗╚██╗ ██╔╝██╔════╝██║ ██╔═══██╗██╔══██╗██╔══██╗ ║ ║ ██████╔╝ ╚████╔╝ █████╗ ██║ ██║ ██║██║ ██║██║ ██║ ║ ║ ██╔═══╝ ╚██╔╝ ██╔══╝ ██║ ██║ ██║██║ ██║██║ ██║ ║ ║ ██║ ██║ ██║ ███████╗╚██████╔╝██████╔╝██████╔╝ ║ ║ ╚═╝ ╚═╝ ╚═╝ ╚══════╝ ╚═════╝ ╚═════╝ ╚═════╝ ║ ║ ║ ║ Steam ID/IP/Email Testing Suite v1.2.5 ║ ║ [Educational Use Only] ║ ╚═══════════════════════════════════════════════════════════════╝ {Colors.END}""" print(banner) def print_menu(self): menu = f""" {Colors.YELLOW}{Colors.BOLD}Available Commands:{Colors.END} {Colors.CYAN}1.{Colors.END} Steam Tools - Configure Steam ID flooding {Colors.CYAN}2.{Colors.END} IP Tools - Configure IP address flooding {Colors.CYAN}3.{Colors.END} Email Tools - Configure Email flooding {Colors.CYAN}4.{Colors.END} Start Attack - Begin configured attack {Colors.CYAN}5.{Colors.END} Stop Attack - Halt current operations {Colors.CYAN}6.{Colors.END} Status - View attack statistics {Colors.CYAN}7.{Colors.END} Save Config - Export current configuration {Colors.CYAN}8.{Colors.END} Load Config - Import saved configuration {Colors.CYAN}9.{Colors.END} Clear Console - Clear terminal output {Colors.CYAN}0.{Colors.END} Exit - Close application {Colors.GREEN}Current Status:{Colors.END} {self.get_status()} """ print(menu) def get_status(self): if self.simulator.active_attacks: return f"{Colors.GREEN}{Colors.BOLD}ATTACKING{Colors.END} ({len(self.simulator.active_attacks)} active)" return f"{Colors.RED}IDLE{Colors.END}" def steam_tools_menu(self): print(f"\n{Colors.BLUE}{Colors.BOLD}=== Steam ID Flooder Configuration ==={Colors.END}\n") targets_input = input(f"{Colors.CYAN}Enter Steam IDs (one per line, empty line to finish):{Colors.END}\n") targets = [] while targets_input.strip(): targets.append(targets_input.strip()) targets_input = input() if not targets: self.logger.error("No targets specified") return None print(f"\n{Colors.YELLOW}Request Type:{Colors.END}") print(" 1. Friend Requests") print(" 2. Message Flood") print(" 3. Profile Comments") choice = input("Select (1-3): ") types = ["Friend Requests", "Message Flood", "Profile Comments"] attack_type = types[int(choice)-1] if choice in ['1','2','3'] else "Friend Requests" count = input(f"\nNumber of attacks per target [50]: ") or "50" message = "" if attack_type == "Message Flood": message = input(f"Custom message [Hello from PyFlood]: ") or "Hello from PyFlood" config = SteamConfig( targets=targets, attack_type=attack_type, count=int(count), message=message ) self.logger.success(f"Steam configuration saved: {len(targets)} targets, {count} requests each") return config def ip_tools_menu(self): print(f"\n{Colors.PURPLE}{Colors.BOLD}=== IP Flooder Configuration ==={Colors.END}\n") targets_input = input(f"{Colors.CYAN}Enter IP addresses (one per line, empty line to finish):{Colors.END}\n") targets = [] while targets_input.strip(): targets.append(targets_input.strip()) targets_input = input() if not targets: self.logger.error("No targets specified") return None print(f"\n{Colors.YELLOW}Attack Type:{Colors.END}") print(" 1. UDP Flood") print(" 2. SYN Flood") print(" 3. HTTP Flood") print(" 4. ICMP Flood") choice = input("Select (1-4): ") types = ["UDP Flood", "SYN Flood", "HTTP Flood", "ICMP Flood"] attack_type = types[int(choice)-1] if choice in ['1','2','3','4'] else "UDP Flood" duration = input(f"\nDuration in seconds [60]: ") or "60" port = 80 if attack_type in ["UDP Flood", "SYN Flood", "HTTP Flood"]: port = input(f"Target port [80]: ") or "80" config = IPConfig( targets=targets, attack_type=attack_type, duration=int(duration), port=int(port) ) self.logger.success(f"IP configuration saved: {len(targets)} targets, {duration}s duration") return config def email_tools_menu(self): print(f"\n{Colors.GREEN}{Colors.BOLD}=== Email Flooder Configuration ==={Colors.END}\n") targets_input = input(f"{Colors.CYAN}Enter email addresses (one per line, empty line to finish):{Colors.END}\n") targets = [] while targets_input.strip(): targets.append(targets_input.strip()) targets_input = input() if not targets: self.logger.error("No targets specified") return None emails_per = input(f"\nEmails per target [20]: ") or "20" delay = input(f"Delay between emails in ms [250]: ") or "250" subject = input(f"Subject line [Important notification]: ") or "Important notification" body = input(f"Email body [Default message]: ") or "This is a test message from PyFlood Suite" config = EmailConfig( targets=targets, emails_per_target=int(emails_per), delay=int(delay), subject=subject, body=body ) self.logger.success(f"Email configuration saved: {len(targets)} targets, {emails_per} emails each") return config def start_attack(self, config): if isinstance(config, SteamConfig): attack_id = f"steam_{int(time.time())}" self.simulator.active_attacks[attack_id] = True thread = threading.Thread( target=self.simulator.steam_flood, args=(config, attack_id) ) thread.daemon = True thread.start() self.logger.info(f"Steam attack thread started: {attack_id}") elif isinstance(config, IPConfig): attack_id = f"ip_{int(time.time())}" self.simulator.active_attacks[attack_id] = True thread = threading.Thread( target=self.simulator.ip_flood, args=(config, attack_id) ) thread.daemon = True thread.start() self.logger.info(f"IP attack thread started: {attack_id}") elif isinstance(config, EmailConfig): attack_id = f"email_{int(time.time())}" self.simulator.active_attacks[attack_id] = True thread = threading.Thread( target=self.simulator.email_flood, args=(config, attack_id) ) thread.daemon = True thread.start() self.logger.info(f"Email attack thread started: {attack_id}") def stop_all_attacks(self): count = len(self.simulator.active_attacks) self.simulator.active_attacks.clear() self.logger.warning(f"Stopped {count} active attack(s)") def save_configuration(self, configs: Dict): try: data = {} for key, config in configs.items(): if config: data[key] = asdict(config) with open(self.config_file, 'w') as f: json.dump(data, f, indent=2) self.logger.success(f"Configuration saved to {self.config_file}") except Exception as e: self.logger.error(f"Failed to save config: {e}") def load_configuration(self): try: if not os.path.exists(self.config_file): self.logger.error("No configuration file found") return {} with open(self.config_file, 'r') as f: data = json.load(f) configs = {} if 'steam' in data: configs['steam'] = SteamConfig(**data['steam']) if 'ip' in data: configs['ip'] = IPConfig(**data['ip']) if 'email' in data: configs['email'] = EmailConfig(**data['email']) self.logger.success("Configuration loaded successfully") return configs except Exception as e: self.logger.error(f"Failed to load config: {e}") return {} def show_status(self): print(f"\n{Colors.BOLD}=== System Status ==={Colors.END}") print(f"Active Attacks: {len(self.simulator.active_attacks)}") print(f"Total Requests Sent: {self.simulator.attack_counter}") print(f"Uptime: {time.strftime('%H:%M:%S', time.gmtime(time.time() - self.start_time))}") print(f"Security: AES-256 Encrypted Connection") print("") def run(self): self.start_time = time.time() current_config = {} self.clear_screen() self.print_banner() self.logger.info("PyFlood Suite v1.2.5 initialized") self.logger.info("Security protocol: AES-256 encrypted") self.logger.info("Ready to initiate flood attacks") self.logger.warning("Use responsibly - Educational purposes only") while self.running: self.print_menu() choice = input(f"{Colors.BOLD}pyflood>{Colors.END} ").strip().lower() if choice in ['1', 'steam', 'steam tools']: config = self.steam_tools_menu() if config: current_config['steam'] = config elif choice in ['2', 'ip', 'ip tools']: config = self.ip_tools_menu() if config: current_config['ip'] = config elif choice in ['3', 'email', 'email tools']: config = self.email_tools_menu() if config: current_config['email'] = config elif choice in ['4', 'start', 'start attack']: if not current_config: self.logger.error("No configuration loaded. Configure tools first.") else: print(f"\n{Colors.YELLOW}Available configurations:{Colors.END}") for i, (key, _) in enumerate(current_config.items(), 1): print(f" {i}. {key.upper()}") print(" 0. ALL") attack_choice = input("Select attack to start: ") if attack_choice == '0': for config in current_config.values(): self.start_attack(config) elif attack_choice.isdigit(): idx = int(attack_choice) - 1 if 0 <= idx < len(current_config): key = list(current_config.keys())[idx] self.start_attack(current_config[key]) elif choice in ['5', 'stop', 'stop attack']: self.stop_all_attacks() elif choice in ['6', 'status']: self.show_status() elif choice in ['7', 'save', 'save config']: self.save_configuration(current_config) elif choice in ['8', 'load', 'load config']: loaded = self.load_configuration() current_config.update(loaded) elif choice in ['9', 'clear', 'clear console']: self.clear_screen() self.print_banner() elif choice in ['0', 'exit', 'quit']: self.running = False self.stop_all_attacks() self.logger.info("Shutting down PyFlood Suite...") time.sleep(1) elif choice == 'help': print(f"\n{Colors.CYAN}Detailed Help:{Colors.END}") print(" This tool simulates network testing scenarios.") print(" Configure targets, select attack vectors, and monitor output.") print(" All traffic is simulated for educational demonstration.") print("") else: self.logger.error(f"Unknown command: {choice}") def main(): parser = argparse.ArgumentParser(description='PyFlood Suite - Network Testing Tool') parser.add_argument('--config', '-c', help='Load configuration file') parser.add_argument('--steam', '-s', action='store_true', help='Steam mode') parser.add_argument('--ip', '-i', action='store_true', help='IP mode') parser.add_argument('--email', '-e', action='store_true', help='Email mode') args = parser.parse_args() suite = PyFloodSuite() if args.config: configs = suite.load_configuration() # Auto-start if config provided pass try: suite.run() except KeyboardInterrupt: print(f"\n{Colors.YELLOW}Interrupted by user{Colors.END}") suite.stop_all_attacks() print(f"{Colors.GREEN}Cleanup complete. Exiting.{Colors.END}") if __name__ == "__main__": main()