基于Paramiko的远程主机SSH服务器代码技术咨询

嘿，我看你写了一段基于Paramiko的SSH服务器代码，目前在密码验证的部分还没写完对吧？我来帮你补全代码、梳理关键要点，顺便解答可能遇到的常见问题：

1. 补全密码验证逻辑 #

你代码里的check_auth_password方法只写了判断条件，还没返回对应的授权结果。Paramiko要求这个方法返回指定的常量，才能让客户端知道验证是否通过。补全后的代码如下：

def check_auth_password(self, username, password):
    if(username=='justin') and (password == 'lovesthepython'):
        return paramiko.AUTH_SUCCESSFUL
    return paramiko.AUTH_FAILED

2. 添加工作Shell的支持 #

仅仅处理会话通道请求还不够，客户端要执行命令的话，还需要支持PTY（伪终端）分配和Shell请求。在你的Server类里添加这两个方法：

def check_channel_shell_request(self, channel):
    self.event.set()
    return True

def check_channel_pty_request(self, channel, term, width, height, pixelwidth, pixelheight, modes):
    return True

这两个方法分别告诉Paramiko：允许客户端申请伪终端，允许开启交互Shell会话。

3. 完整的服务器运行逻辑 #

你还需要添加主程序代码来监听连接、处理客户端请求，以及执行命令并返回结果。完整的代码整合如下：

import threading
import paramiko
import subprocess
import sys
import socket

host_key = paramiko.RSAKey(filename='test_rsa.key')

class Server(paramiko.ServerInterface):
    def __init__(self):
        self.event = threading.Event()

    def check_channel_request(self, kind, chanid):
        if kind == 'session':
            return paramiko.OPEN_SUCCEEDED
        return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED

    def check_auth_password(self, username, password):
        if(username=='justin') and (password == 'lovesthepython'):
            return paramiko.AUTH_SUCCESSFUL
        return paramiko.AUTH_FAILED

    def check_channel_shell_request(self, channel):
        self.event.set()
        return True

    def check_channel_pty_request(self, channel, term, width, height, pixelwidth, pixelheight, modes):
        return True

def handle_client(client):
    try:
        transport = paramiko.Transport(client)
        transport.add_server_key(host_key)
        server = Server()
        try:
            transport.start_server(server=server)
        except paramiko.SSHException:
            print("SSH协商失败，无法建立连接")
            return
        # 等待客户端发起会话请求
        channel = transport.accept(20)
        if not channel:
            print("没有打开任何通道，连接终止")
            return
        server.event.wait(10)
        if not server.event.is_set():
            print("客户端未请求Shell会话")
            return
        # 向客户端发送欢迎信息
        channel.send("欢迎来到自定义SSH服务器！\r\n")
        while True:
            channel.send("$ ")
            # 接收客户端输入的命令
            command = b""
            while b"\r\n" not in command:
                command += channel.recv(1024)
            command = command.strip().decode('utf-8')
            # 处理退出命令
            if command.lower() == 'exit':
                channel.send("再见！\r\n")
                break
            # 执行命令并返回结果
            try:
                result = subprocess.run(command, shell=True, capture_output=True, text=True)
                output = result.stdout + result.stderr
                channel.send(output + "\r\n")
            except Exception as e:
                channel.send(f"执行命令出错: {str(e)}\r\n")
        channel.close()
    except Exception as e:
        print(f"处理客户端请求时出错: {str(e)}")
    finally:
        client.close()

if __name__ == "__main__":
    # 创建监听Socket
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    # 监听本地2222端口（避免和系统默认SSH端口22冲突）
    server_socket.bind(('0.0.0.0', 2222))
    server_socket.listen(100)
    print("正在2222端口监听SSH连接...")
    try:
        while True:
            client, addr = server_socket.accept()
            print(f"接受来自 {addr[0]}:{addr[1]} 的连接")
            # 启动线程处理客户端请求
            threading.Thread(target=handle_client, args=(client,)).start()
    except KeyboardInterrupt:
        print("服务器正在关闭")
        server_socket.close()
        sys.exit(0)

4. 生成所需的RSA密钥文件 #

你代码里用到了test_rsa.key，需要先生成这个密钥文件。可以用ssh-keygen命令：

ssh-keygen -t rsa -f test_rsa.key -N ""

或者用Python代码生成：

paramiko.RSAKey.generate(bits=2048).write_private_key_file('test_rsa.key')

没有这个密钥文件的话，服务器启动时会直接报错，因为Paramiko需要它来完成SSH握手的密钥交换。

常见问题排查 #

• 连接被拒绝：检查服务器是否监听了正确的端口，防火墙是否允许该端口的入站流量。
• 密码验证失败：确认用户名和密码的判断逻辑正确，返回的是paramiko.AUTH_SUCCESSFUL/paramiko.AUTH_FAILED常量，而不是普通的布尔值。
• 无法执行命令：确保已经实现了check_channel_shell_request和check_channel_pty_request方法，并且在handle_client函数中正确处理了命令执行和结果返回。

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