AsyncSSH基础SFTP服务器权限错误：连接失败

问题描述 #

尝试用Python的AsyncSSH库实现单元测试用的基础SFTP服务器，需求是自定义根路径并设置用户名/密码验证，但运行代码时出现权限错误。

原代码：

import asyncio
import asyncssh
import os
import logging

logging.basicConfig(level=logging.DEBUG)

class BasicSFTPServer(asyncssh.SFTPServer):
    def __init__(self):
        super().__init__(chan=22, chroot=os.path.basename(__file__))

async def main():
    host_key = asyncssh.generate_private_key("ssh-rsa")
    server = await asyncssh.listen(
        host="127.0.0.1",
        port=0,
        server_host_keys=[host_key],
        sftp_factory=BasicSFTPServer
    )
    server_port = server.get_port()

    async with asyncssh.connect(host="localhost", port=server_port, known_hosts=None) as conn:
        async with conn.start_sftp_client() as sftp:
            result = await sftp.listdir()
            print(result)

    server.close()
    await server.wait_closed()

if(__name__ == "__main__"):
    asyncio.run(main())

运行时错误信息：

DEBUG:asyncio:Using proactor: IocpProactor
INFO:asyncssh:Creating SSH listener on 127.0.0.1
INFO:asyncssh:Host canonicalization disabled
INFO:asyncssh:Opening SSH connection to localhost, port 53559
INFO:asyncssh:[conn=0] Connected to SSH server at localhost, port 53559
INFO:asyncssh:[conn=0]   Local address: 127.0.0.1, port 53561
INFO:asyncssh:[conn=0]   Peer address: 127.0.0.1, port 53559
DEBUG:asyncssh:[conn=0] Sending version SSH-2.0-AsyncSSH_2.21.0
INFO:asyncssh:[conn=1] Accepted SSH client connection
INFO:asyncssh:[conn=1]   Local address: 127.0.0.1, port 53559
INFO:asyncssh:[conn=1]   Peer address: 127.0.0.1, port 53561
DEBUG:asyncssh:[conn=1] Sending version SSH-2.0-AsyncSSH_2.21.0
DEBUG:asyncssh:[conn=1] Received version SSH-2.0-AsyncSSH_2.21.0
DEBUG:asyncssh:[conn=1] Requesting key exchange
DEBUG:asyncssh:[conn=0] Received version SSH-2.0-AsyncSSH_2.21.0
DEBUG:asyncssh:[conn=0] Requesting key exchange
DEBUG:asyncssh:[conn=0] Received key exchange request
DEBUG:asyncssh:[conn=0] Beginning key exchange
DEBUG:asyncssh:[conn=1] Received key exchange request
DEBUG:asyncssh:[conn=1] Beginning key exchange
DEBUG:asyncssh:[conn=0] Completed key exchange
DEBUG:asyncssh:[conn=1] Completed key exchange
INFO:asyncssh:[conn=0] Beginning auth for user Manuel
INFO:asyncssh:[conn=1] Beginning auth for user Manuel
INFO:asyncssh:[conn=0] Auth failed for user Manuel
INFO:asyncssh:[conn=0] Connection failure: Permission denied for user Manuel on host localhost
INFO:asyncssh:[conn=0] Aborting connection
Traceback (most recent call last):
  File "...\Documents\workspace\repos\template-python-service\debug\main.py", line 34, in <module>
    asyncio.run(main())
    ~~~~~~~~~~~^^^^^^^^
  File "...\AppData\Roaming\uv\python\cpython-3.13.1-windows-x86_64-none\Lib\asyncio\runners.py", line 194, in run
    return runner.run(main)
           ~~~~~~~~~~^^^^^^
  File "...\AppData\Roaming\uv\python\cpython-3.13.1-windows-x86_64-none\Lib\asyncio\runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "...\AppData\Roaming\uv\python\cpython-3.13.1-windows-x86_64-none\Lib\asyncio\base_events.py", line 720, in run_until_complete
    return future.result()
           ~~~~~~~~~~~~~^^
  File "...\Documents\workspace\repos\template-python-service\debug\main.py", line 25, in main
    async with asyncssh.connect(host="localhost", port=server_port, known_hosts=None) as conn:
               ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "...\Documents\workspace\repos\template-python-service\.venv\Lib\site-packages\asyncssh\misc.py", line 386, in __aenter__
    self._coro_result = await self._coro
                        ^^^^^^^^^^^^^^^^
  File "...\Documents\workspace\repos\template-python-service\.venv\Lib\site-packages\asyncssh\connection.py", line 9186, in connect
    return await asyncio.wait_for(
           ^^^^^^^^^^^^^^^^^^^^^^^
    ...<2 lines>...
        timeout=new_options.connect_timeout)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "...\AppData\Roaming\uv\python\cpython-3.13.1-windows-x86_64-none\Lib\asyncio\tasks.py", line 507, in wait_for
    return await fut
           ^^^^^^^^^
  File "...\.venv\Lib\site-packages\asyncssh\connection.py", line 528, in _connect
    await options.waiter
asyncssh.misc.PermissionDenied: Permission denied for user xxxx on host localhost

错误原因 #

1. 缺少用户认证配置：AsyncSSH默认没有启用任何用户认证方式，导致客户端连接时被拒绝。
2. SFTPServer初始化参数错误：__init__方法中硬编码chan=22不符合AsyncSSH的参数要求，且chroot路径使用os.path.basename(__file__)会指向当前脚本文件名而非目录，无法作为SFTP根路径。

修复后的完整代码 #

import asyncio
import asyncssh
import os
import logging

logging.basicConfig(level=logging.DEBUG)

# 自定义用户密码认证回调
def password_auth_handler(username, password):
    # 这里设置测试用的用户名和密码
    allowed_users = {"test_user": "test_pass"}
    return allowed_users.get(username) == password

class BasicSFTPServer(asyncssh.SFTPServer):
    def __init__(self, chan, ctx):
        # 设置自定义根路径，这里用当前脚本所在目录作为示例
        root_path = os.path.dirname(os.path.abspath(__file__))
        super().__init__(chan, ctx, chroot=root_path)

async def main():
    host_key = asyncssh.generate_private_key("ssh-rsa")
    server = await asyncssh.listen(
        host="127.0.0.1",
        port=0,
        server_host_keys=[host_key],
        sftp_factory=BasicSFTPServer,
        # 启用密码认证并指定回调函数
        password_auth_handler=password_auth_handler
    )
    server_port = server.get_port()

    # 客户端连接时指定用户名和密码
    async with asyncssh.connect(
        host="localhost",
        port=server_port,
        known_hosts=None,
        username="test_user",
        password="test_pass"
    ) as conn:
        async with conn.start_sftp_client() as sftp:
            result = await sftp.listdir()
            print("SFTP根目录内容:", result)

    server.close()
    await server.wait_closed()

if __name__ == "__main__":
    asyncio.run(main())

代码说明 #

• 用户认证配置：添加password_auth_handler回调函数，定义允许登录的用户名和密码，并在asyncssh.listen中指定该回调启用密码认证。
• 修复SFTPServer初始化：__init__方法接收chan和ctx参数（AsyncSSH会自动传入），使用os.path.dirname(os.path.abspath(__file__))获取当前脚本所在目录作为SFTP根路径，也可以替换为任意你需要的测试目录。
• 客户端连接调整：在asyncssh.connect中添加username和password参数，使用配置好的测试账号登录。

内容的提问来源于stack exchange，提问作者Manuel Díaz Pozo