Python Paramiko SSH服务器PTY分配失败及连接断开问题排查

1. 补全Paramiko服务器的PTY处理逻辑 #

多数参考代码未实现PTY分配回调，导致客户端PTY请求被直接拒绝，触发后续Shell失败。需重载SSHServerInterface的两个关键方法：

class SSHProxyServer(paramiko.ServerInterface):
    def check_channel_request(self, kind, chanid):
        # 允许会话通道请求
        if kind == 'session':
            return paramiko.OPEN_SUCCEEDED
        return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED

    def get_pty(self, channel, term, width, height, pixelwidth, pixelheight, modes):
        # 显式同意PTY分配请求，否则客户端会报PTY失败
        return True

2. 实现Shell会话的双向转发逻辑 #

客户端发起Shell请求后，必须在中间人服务器与真实SSH服务器之间建立通道并双向转发数据，否则Shell请求会失败且通道提前关闭：

class SSHProxyServer(paramiko.ServerInterface):
    def __init__(self):
        self.event = threading.Event()
        self.remote_conn = None
        self.remote_channel = None

    def check_channel_shell(self, channel):
        # 连接到真实SSH服务器
        self.remote_conn = paramiko.SSHClient()
        self.remote_conn.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.remote_conn.connect('真实服务器IP', username='tim', password='对应密码')
        
        # 开启远程Shell通道
        self.remote_channel = self.remote_conn.invoke_shell()
        
        # 启动双向数据转发线程
        threading.Thread(target=self.forward_local_to_remote, args=(channel, self.remote_channel)).start()
        threading.Thread(target=self.forward_remote_to_local, args=(self.remote_channel, channel)).start()
        
        return True

    def forward_local_to_remote(self, local_chan, remote_chan):
        try:
            while True:
                data = local_chan.recv(1024)
                if not data:
                    break
                print(f"→ 客户端发送: {repr(data)}")
                remote_chan.send(data)
        except Exception as e:
            print(f"本地→远程转发错误: {e}")
        finally:
            local_chan.close()
            remote_chan.close()

    def forward_remote_to_local(self, remote_chan, local_chan):
        try:
            while True:
                data = remote_chan.recv(1024)
                if not data:
                    break
                print(f"← 服务器返回: {repr(data)}")
                local_chan.send(data)
        except Exception as e:
            print(f"远程→本地转发错误: {e}")
        finally:
            local_chan.close()
            remote_chan.close()

3. 修正Socket生命周期管理 #

确保服务器端Socket不会提前关闭，需在会话结束后再释放资源：

def handle_client(client_socket):
    try:
        transport = paramiko.Transport(client_socket)
        host_key = paramiko.RSAKey.from_private_key_file('server.key')
        transport.add_server_key(host_key)
        
        server = SSHProxyServer()
        transport.start_server(server=server)
        
        # 等待客户端认证并建立通道
        channel = transport.accept(30)
        if not channel:
            print("客户端认证超时")
            return
        
        # 等待会话结束事件，避免提前关闭Socket
        server.event.wait()
    except Exception as e:
        print(f"客户端处理错误: {e}")
    finally:
        client_socket.close()

4. 检查环境权限与日志 #

• 确认运行脚本的用户有权限访问/dev/pts，执行ls -l /dev/pts验证权限（通常归属root或tty组）；
• 开启Paramiko详细日志定位细节：

  import logging
  logging.basicConfig(level=logging.DEBUG)
  paramiko.util.log_to_file('paramiko_debug.log')
  

  查看日志中PTY请求处理、通道关闭的具体触发点。

内容的提问来源于stack exchange，提问作者Marco Montevechi Filho