Paramiko与ssh2-python3高线程场景下运行异常排查

我开发了一款用于研究学习的SSH密码校验工具，目的是深入理解线程与安全相关细节，无恶意用途。但在高线程（200、500甚至1000线程）场景下遇到连接数量限制问题，分别使用Paramiko和ssh2-python3库均出现异常，以下是问题详情、原因分析和解决方法。

一、问题现象 #

1. Paramiko库异常 #

线程数达到12及以上时，出现paramiko.ssh_exception.SSHException: Error reading SSH protocol banner错误，仅少量匹配结果可靠，整体稳定性极差。

2. ssh2-python3库异常 #

线程数达到50-60左右后，工具停止返回成功匹配结果，线程数越高结果一致性越差；改为连接多IP后情况略有改善，但仍无法100%可靠运行。

二、核心疑问 #

1. 这类异常的产生原因是什么？
2. 有哪些可行的解决方法？

三、原因分析 #

1. 系统级套接字限制 #

Linux/Windows系统默认对单进程可打开的文件描述符（套接字属于文件描述符）数量有限制，高线程场景下大量并发创建套接字会触发这个限制，导致连接失败或Banner读取超时。

2. SSH服务器的并发连接限制 #

大多数SSH服务器（如OpenSSH）默认会限制同一IP的并发连接数，防止暴力破解。高线程并发请求会被服务器识别为恶意行为，主动拒绝或延迟响应，导致Banner读取失败、认证无响应。

3. 库本身的线程安全与资源竞争问题 #

• Paramiko：部分底层网络操作未做完善的线程同步，高线程下可能出现套接字资源竞争，导致Banner读取时数据混乱。
• ssh2-python3：虽然基于libssh2（本身线程安全），但手动创建套接字时未处理好资源释放，高线程下套接字泄漏或超时设置不合理，导致连接失败。

4. 线程池过载与超时设置不合理 #

线程数过高会导致系统上下文切换频繁，网络请求的响应时间被拉长，超出设置的超时时间（如Paramiko的banner_timeout、ssh2-python3的sock.settimeout），最终触发异常。

四、可行解决方法 #

1. 调整系统资源限制 #

• Linux系统：临时调整当前进程的文件描述符限制：ulimit -n 4096；永久修改可编辑/etc/security/limits.conf，添加* soft nofile 4096和* hard nofile 8192。
• Windows系统：通过组策略或注册表调整最大套接字连接数（路径：HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Services\Tcpip\Parameters，修改MaxUserPort和TcpTimedWaitDelay）。

2. 控制并发线程数，采用动态线程池或异步IO #

• 放弃固定高线程数，改为动态调整线程池大小，比如基于目标服务器的负载或当前系统资源使用情况调整max_workers。
• 改用异步IO框架（如asyncssh）替代多线程，异步模型能更高效处理大量并发连接，避免线程上下文切换开销。

3. 优化SSH库的使用方式 #

• Paramiko：
  • 增加banner_timeout时长，同时设置auth_timeout参数避免认证超时。
  • 复用SSH连接（如果目标是同一服务器），减少重复创建连接的开销。
  • 确保每个连接在异常时都被正确关闭，避免资源泄漏：在finally块中调用cli.close()。
• ssh2-python3：
  • 在finally块中关闭套接字和Session，释放资源：

    finally:
        sock.close()
        session.disconnect()
    

  • 增加握手和认证的超时时间，避免因系统负载高导致的超时。

4. 模拟真实请求行为，规避服务器限制 #

• 给每个线程的请求添加随机延迟（如time.sleep(random.uniform(0.1, 0.5))），避免短时间内大量请求集中发送到同一服务器。
• 采用多IP代理（如果有条件），分散请求来源，绕过服务器的单IP并发限制。

5. 增加错误重试机制 #

对连接失败、Banner读取失败的请求进行有限次数的重试，比如重试2-3次，提高匹配结果的可靠性。

五、可复现的最小示例代码 #

Paramiko示例代码 #

import random
from concurrent.futures import ThreadPoolExecutor
from itertools import product

import paramiko


def paramiko_core(server, port, username, password):
    cli = paramiko.SSHClient()
    cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        cli.connect(
            server,
            port,
            username,
            password,
            banner_timeout=150,
        )
        cli.close()
    except paramiko.SSHException:
        ...


# Put some dummy or safe ips in the file (I put about 40 or so)
with open("sv_list.txt") as f:
    list_ips = f.readlines()
# And put a valid ip address here
list_ips.insert(20, "127.0.0.1")
# Replace a valid username here...
list_username = ["user1", "user2", "real_user", "user3", "user4", "user5"]
# Replace a valid password here...
list_password = ["pass1", "pass2", "real_pass", "pass3", "pass4", "pass5"]

with ThreadPoolExecutor(max_workers=16) as executor:
    comb_list = list(product(list_ips, list_username, list_password))
    random.shuffle(comb_list)
    for ip, username, password in comb_list:
        executor.submit(paramiko_core, ip.strip(), 22, username, password)

ssh2-python3示例代码 #

import random
import socket
from concurrent.futures import ThreadPoolExecutor
from itertools import product

from ssh2.exceptions import (
    AuthenticationError,
    SocketDisconnectError,
    SocketRecvError,
    Timeout,
)
from ssh2.session import Session


def ssh2_core(server, port, username, password):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(4)

    try:
        sock.connect((server, port))
    except OSError:
        return False

    session = Session()
    session.set_timeout(1000)

    try:
        session.handshake(sock)
        session.userauth_password(username, password)
        session.open_session()
    except Timeout:
        return False
    except SocketRecvError:
        return False
    except SocketDisconnectError:
        return False
    except AuthenticationError:
        return False

    return True


# Put some dummy or safe ips in the file (I put about 40 or so)
with open("sv_list.txt") as f:
    list_ips = f.readlines()
# And put a valid ip address here
list_ips.insert(20, "127.0.0.1")

list_username = [
    *[f"user{i}" for i in range(3)],
    "real_user",  # Replace a valid username here...
    *[f"user{i}" for i in range(3)],
]

list_password = [
    *[f"pass{i}" for i in range(3)],
    "real_pass",  # Replace a valid password here...
    *[f"pass{i}" for i in range(3)],
]

result = []
with ThreadPoolExecutor(max_workers=120) as executor:
    comb_list = list(product(list_ips, list_username, list_password))
    random.shuffle(comb_list)
    for ip, username, password in comb_list:
        result.append(executor.submit(ssh2_core, ip.strip(), 22, username, password))

if True in [rslt.result() for rslt in result]:
    print("This time it worked!")
else:
    print("This time it failed!")

内容的提问来源于stack exchange，提问作者D3F4U1T