AsyncSSH Basic SFTP Server permission error - connection fails

Question

I'm trying to implement a very basic sftp server in python with asyncssh to be used during a unittest. I want to use a custom path to set the root of the sftp server and it would be better to set a username/password.

So far I have this:

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())

But I get a permission error:

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

I know I'm missing something but I can't find any example...

Where in your code are credentials being handled? Something is using "Manuel" as ssh username. — robertklep
– robertklep, Commented Jul 31 at 17:35
Why are you listening on 127.0.0.1, but connecting to localhost? Be consistent. — Martin Prikryl
– Martin Prikryl, Commented Aug 1 at 6:33
Also your code does not look like the AsyncSSH server examples do. Why? asyncssh.readthedocs.io/en/latest/#server-examples — Martin Prikryl
– Martin Prikryl, Commented Aug 1 at 6:35
@robertklep the script does not contains any credentials yet because I cannot find an example. Manuel is the name of the system user from which I'm executing the script, it seems it is automatically filled by the module as it does not appear anywhere in the script. — Manuel Díaz Pozo
– Manuel Díaz Pozo, Commented Aug 1 at 12:26
I found github.com/phenobarbital/aiosftp/blob/main/aiosftp/scp.py It seems that you need an SSHServer to manage authentication and other stuff and the SFTP server for the main functionality. — Manuel Díaz Pozo
– Manuel Díaz Pozo, Commented Aug 1 at 14:22

Manuel Díaz Pozo · Accepted Answer · 2025-08-02 09:25:09Z

-1

The simplest solution:

import asyncssh
import os
import functools

class SSHServerHandler(asyncssh.SSHServer):
    def connection_made(self, conn) -> None:
        pass

    def connection_lost(self, exc) -> None:
        pass
    
    async def begin_auth(self, username:str) -> bool:
        return True

    def password_auth_supported(self) -> bool:
        return True

    def kbdint_auth_supported(self) -> bool:
        return False

    def public_key_auth_supported(self) -> bool:
        return False

    async def validate_password(self, username:str, password:str) -> bool:
        return username == "user" and password == "pass"

    def auth_completed(self, **kwargs) -> bool:
        return True

class SFTPServerHandler(asyncssh.SFTPServer):
    def __init__(self, chan:asyncssh.SSHServerChannel, root:str):
        os.makedirs(root, exist_ok=True)
        super().__init__(chan, chroot=root)

class SFTPServer:
    def __init__(self, host:str, port:int, path:str):
        self.server = None
        self.host = host
        self.port = port
        self.path = path
        self.host_key = asyncssh.generate_private_key("ssh-rsa")

    async def start(self):
        self.server = await asyncssh.listen(
            server_factory=SSHServerHandler,
            sftp_factory=functools.partial(SFTPServerHandler, root=self.path),
            host=self.host,
            port=self.port,
            server_host_keys=[self.host_key]
        )
        self.port = self.server.get_port()

    async def close(self):
        self.server.close()
        await self.server.wait_closed()

Tested and working

answered Aug 2 at 9:25

Manuel Díaz Pozo

1

Sign up to request clarification or add additional context in comments.

1 Comment

Community Aug 2 at 10:28

Your answer could be improved with additional supporting information. Please edit to add further details, such as citations or documentation, so that others can confirm that your answer is correct. You can find more information on how to write good answers in the help center.

Collectives™ on Stack Overflow

AsyncSSH Basic SFTP Server permission error - connection fails

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related