创建一个简单的SSH服务器

发表于 2022-07-27 更新于 2026-02-23 分类于 Linux 阅读次数：

0x00 前言

为了加深对SSH协议的理解，准备自己实现一个SSH服务端，需要同时支持Windows、Linux、MacOS三大系统。为了尽量提升性能，准备使用协程(asyncio)来开发。

0x01 基于AsyncSSH开发一个最简单的SSH服务端

在调研了几个开源的python SSH库后，最终选择了AsyncSSH。这个库基于asyncio开发，符合我们的要求，同时扩展性也比较好。

下面实现了一个使用固定账号密码登录的SSH服务器，登录成果后会打印一串字符串，并退出：

import asyncio
import asyncssh

async def start_ssh_server():
    def handle_client(process):
        process.stdout.write("Welcome to my SSH server, byebye!\n")
        process.exit(0)

    class MySSHServer(asyncssh.SSHServer):
        def __init__(self):
            self._conn = None

        def password_auth_supported(self):
            return True

        def validate_password(self, username, password):
            return username == "drunkdream" and password == "123456"

        def connection_made(self, conn):
            print("Connection created", conn.get_extra_info("peername")[0])
            self._conn = conn

        def connection_lost(self, exc):
            print("Connection lost", exc)

    await asyncssh.create_server(
        MySSHServer,
        "",
        2222,
        server_host_keys=["skey"],
        process_factory=handle_client,
    )

    await asyncio.sleep(1000)

loop = asyncio.get_event_loop()
loop.run_until_complete(start_ssh_server())

server_host_keys是服务端的私钥文件列表，用于在建立连接时验证服务端的合法性；在第一次连接时客户端会弹出验证指纹的提示，选择yes后会将指纹保存到本地，下次连接时会验证指纹是否匹配，不匹配会报错。

1
2
3

The authenticity of host '[127.0.0.1]:2222 ([127.0.0.1]:2222)' can't be established.
RSA key fingerprint is SHA256:nyXXvfYgedKWPRnhl1ss6k+R5cqFleUQu/fDhYYXESI.
Are you sure you want to continue connecting (yes/no)?

ssh drunkdream@127.0.0.1 -p 2222
Password:
Welcome to my SSH server, byebye!
Connection to 127.0.0.1 closed.

这样就实现了一个最简单的SSH服务器了，由此可见，使用AsyncSSH开发SSH服务端是非常方便的。

0x02 支持Shell命令

SSH最常用的功能就是远程终端（shell），下面来实现一个支持执行命令的SSH服务：

async def start_ssh_server():
    import asyncssh

    async def handle_client(process):
        proc = await asyncio.create_subprocess_shell(
            process.command or "bash -i",
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            close_fds=True,
        )
        stdin = proc.stdin
        stdout = proc.stdout
        stderr = proc.stderr

        tasks = [None, None, None]

        while proc.returncode is None:
            if tasks[0] is None:
                tasks[0] = asyncio.ensure_future(process.stdin.read(4096))
            if tasks[1] is None:
                tasks[1] = asyncio.ensure_future(stdout.read(4096))
            if tasks[2] is None:
                tasks[2] = asyncio.ensure_future(stderr.read(4096))

            done_tasks, _ = await asyncio.wait(
                tasks, return_when=asyncio.FIRST_COMPLETED
            )

            for task in done_tasks:
                index = tasks.index(task)
                assert index >= 0
                tasks[index] = None
                buffer = task.result()
                if not buffer:
                    return -1

                if index == 0:
                    stdin.write(buffer)
                elif index == 1:
                    process.stdout.write(buffer.replace(b"\n", b"\r\n"))
                else:
                    process.stderr.write(buffer.replace(b"\n", b"\r\n"))
        return proc.returncode

    class MySSHServer(asyncssh.SSHServer):
        def __init__(self):
            self._conn = None

        def password_auth_supported(self):
            return True

        def validate_password(self, username, password):
            return username == "drunkdream" and password == "123456"

        def connection_made(self, conn):
            print("Connection created", conn.get_extra_info("peername")[0])
            self._conn = conn

        def connection_lost(self, exc):
            print("Connection lost", exc)

    await asyncssh.create_server(
        MySSHServer,
        "",
        2222,
        server_host_keys=["skey"],
        process_factory=lambda process: asyncio.ensure_future(handle_client(process)),
        encoding=None,
        line_editor=False
    )

    await asyncio.sleep(1000)

与前一个版本相比，主要是修改了handle_client实现，变成了一个协程函数，里面创建了子进程，并支持将ssh客户端输入的命令传给子进程，然后将子进程的stdout和stderr转发给ssh客户端。注意到，这里将line_editor参数设置成了False，主要是为了支持实时命令交互。这个参数后面还会详细介绍。

上面的代码在实际使用中发现，对于很快执行完的命令，如：ifconfig等，使用上没什么问题，但是如果输入python命令进入交互式界面，就会卡住没有任务输入。这是因为使用create_subprocess_shell方式创建的子进程不支持pty导致的。

0x03 支持pty

pty(pseudo-tty)是伪终端的意思，也就是虚拟了一个终端出来，让进程可以像正常终端一样进行交互（通常情况下通过管道重定向输入输出的进程都无法支持交互式操作）。交互式终端下缓冲模式是无缓冲（字符模式），也就是stdout每次只要有输出就会打印出来；而非交互式终端是行缓冲模式，stdout必须收到\n换行符才会打印出来。

也就是说，如果终端要支持像python交互式命令这样的场景，必须支持pty。python中可以通过sys.stdout.isatty()来判断当前进程是否支持伪终端。

python -c 'import sys;print(sys.stdout.isatty())'
True

python -c 'import sys;print(sys.stdout.isatty())' > /tmp/1.txt && cat /tmp/1.txt
False

python -c 'import pty; pty.spawn(["python", "-c", "import sys;print(sys.stdout.isatty())"])' > /tmp/1.txt && cat /tmp/1.txt
True

从上面可以看出，经过重定向之后，isatty返回值变成了False；但是使用pty.spawn函数之后，重定向就不会影响isatty的返回值了。这里的秘密就在于pty库实现了一个虚拟的tty，具体实现原理我们后面有时间再来分析。

因此，可以使用以下代码创建一个支持pty的子进程：

import pty

cmdline = list(shlex.split(command or os.environ.get("SHELL", "sh")))
exe = cmdline[0]
if exe[0] != "/":
    for it in os.environ["PATH"].split(":"):
        path = os.path.join(it, exe)
        if os.path.isfile(path):
            exe = path
            break

pid, fd = pty.fork()
if pid == 0:
    # child process
    sys.stdout.flush()
    try:
        os.execve(exe, cmdline, os.environ)
    except Exception as e:
        sys.stderr.write(str(e))
else:
    # parent process
    print(os.read(fd, 4096))

上面的方法只能支持Linux和MacOS系统，Windows 1809以上版本可以使用以下方法：

cmd = (
    "conhost.exe",
    "--headless",
    "--width",
    str(size[0]),
    "--height",
    str(size[1]),
    "--",
    command or "cmd.exe",
)
proc = await asyncio.create_subprocess_exec(
    *cmd,
    stdin=asyncio.subprocess.PIPE,
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
)

conhost.exe里使用CreatePseudoConsole等相关函数，实现了伪终端。低版本Windows上就需要使用其它方式来支持了，例如：winpty。

0x04 行编辑器模式

前面提到，在使用asyncssh.create_server函数创建SSH服务端时，有个line_editor参数设置成了False。这表示关闭了行编辑器模式，也就是说任何输入的字符都会被实时发送给shell进程，一般这种都是shell进程拥有伪终端的情况。

但如果创建的是一个不支持伪终端的shell进程，就必须关闭行编辑器模式，也就是将line_editor置为True。此时，SSH客户端输入的字符会被asyncssh库捕获并进行处理，直到用户按下Enter键的时候，才会将输入一次性发送给shell进程。

具体可以参考文档。

0x05 支持端口转发

SSH服务器有个非常有用的功能就是端口转发，包括正向端口转发和反向端口转发。使用方法如下：

正向端口转发：

1	ssh -L 127.0.0.1:7778:127.0.0.1:7777 root@1.2.3.4

此时，可以将远程机器上的7777端口映射到本地的7778端口。

反向端口转发：

1	ssh -R 127.0.0.1:7778:127.0.0.1:7777 root@1.2.3.4

此时，可以将本地的7777端口映射到远程机器上的7778端口。

要支持端口转发，只需要MySSHServer类增加connection_requested和server_requested方法即可。

async def connection_requested(self, dest_host, dest_port, orig_host, orig_port):
    # 正向端口转发
    return await self._conn.forward_connection(dest_host, dest_port)

def server_requested(self, listen_host, listen_port):
    # 反向端口转发
    return True

0x06 支持密钥登录

通常我们登录SSH服务器，更多的是使用密钥方式登录。要开启这个特性只需要增加以下两个方法即可：

def public_key_auth_supported(self):
    return True

def validate_public_key(self, username, key):
    return True

0x07 总结

使用AsyncSSH库开发SSH服务器还是比较简单的，很多特性都已经封装好了，只要重写一下对应的方法，返回True就可以了。同时，它也提供了高级可定制化的能力，以便实现较为复杂的功能。

完整的SSH服务器代码可以参考：https://github.com/drunkdream/turbo-tunnel/blob/master/turbo_tunnel/ssh.py#L24。