如何用Paramiko维持SSH连接执行命令？代码连接状态判断异常

核心问题 #

你实例化Host对象后，没有调用connect()方法建立SSH连接，直接调用isconnected()时，self.ssh_client仍然是初始化时的None，所以会显示未连接。同时当前isconnected()方法返回对象或None的设计不符合"判断连接状态"的语义，应该返回布尔值更合理。

修正后的代码 #

完善后的Host类

import paramiko


class Host:
    """SSH连接管理类，负责建立连接、执行命令、断开连接"""

    def __init__(self, remote_host, login, password):
        self.remote_host = remote_host
        self.login = login
        self.password = password
        self.ssh_client = None

    def connect(self):
        try:
            self.ssh_client = paramiko.SSHClient()
            self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            self.ssh_client.connect(
                self.remote_host,
                username=self.login,
                password=self.password,
                timeout=1
            )
            return "Connected to SSH!"
        except paramiko.ssh_exception.SSHException as ssh_exception:
            return f"SSH连接失败: {ssh_exception}"
        except TimeoutError as timeout_error:
            return f"主机超时不可用: {timeout_error}"

    def isconnected(self):
        # 修正：返回布尔值判断真实连接状态
        if self.ssh_client:
            transport = self.ssh_client.get_transport()
            return transport is not None and transport.is_active()
        return False

    def exec_pwd(self):
        # 新增：执行pwd命令并返回结果
        if not self.isconnected():
            return "未建立SSH连接，无法执行命令"
        try:
            stdin, stdout, stderr = self.ssh_client.exec_command('pwd')
            stdout_content = stdout.read().decode().strip()
            stderr_content = stderr.read().decode().strip()
            if stderr_content:
                return f"命令执行错误: {stderr_content}"
            return f"当前远程工作目录: {stdout_content}"
        except Exception as e:
            return f"执行命令时出错: {e}"

    def disconnect(self):
        if self.ssh_client is not None:
            self.ssh_client.close()
            self.ssh_client = None  # 重置客户端状态，保证后续状态判断准确
            return "已断开SSH连接!"
        else:
            return "没有活跃的SSH连接。"

完整流程测试代码

print("测试SSH连接流程...")
host = Host("192.168.1.10", 'administrator', 'SuperStrongP477')

# 1. 建立连接
connect_result = host.connect()
print(connect_result)

# 2. 检查连接状态并执行pwd命令
if host.isconnected():
    print("SSH已连接。")
    pwd_result = host.exec_pwd()
    print(pwd_result)
else:
    print("SSH未连接。")

# 3. 断开连接
disconnect_result = host.disconnect()
print(disconnect_result)

关键修改说明 #

1. 必须调用connect()：实例化对象后，必须先调用connect()触发连接逻辑，self.ssh_client才会被赋值为有效的SSH客户端实例。
2. 优化连接状态判断：isconnected()不仅检查客户端实例是否存在，还通过get_transport().is_active()验证连接是否真的处于活跃状态，避免实例存在但连接已断开的误判。
3. 新增pwd命令执行逻辑：专门实现exec_pwd()方法，包含连接状态校验和错误处理，确保命令执行的可靠性。
4. 断开连接后重置状态：关闭连接后将self.ssh_client设为None，保证后续状态判断的准确性。

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