"puttygen: Inappropriate ioctl for device" error and pty problems.

lastochka364 commented 1 year ago

Hello to everyone who is reading this!

First of all, I want to thank @ronf for such an amazing project, as asyncssh! All your work is an unrealistic scale for me, but I'm still very pleasured in using this tool, and just to say - it's very well-documented, special thanks for this! :)

I will use the third person form of communication, as I believe that the application described here, and hopefully, the solutions to the problems may be useful for others. Warning: in this issue will be many questions, and even more code, but I'll try my best to fully describe the problems I encountered and also highlight them all.

Task and problems

Task: generate an SSH key on a remote host using the puttygen command.

Problems:

while trying to generate the .ppk key with passing the key password two times in interactive mode, in a such way as

proc = await conn.create_process()
proc.stdin.write('puttygen -q -t rsa -b 4096 -o ~/mykey.ppk')

or in non-interactive, like this

await conn.create_process('puttygen -q -t rsa -b 4096 -o ~/mykey.ppk')

i'm getting the next error for both cases:

Enter passphrase to save key: 
Re-enter passphrase to verify: 
puttygen: unable to read new passphrase: Inappropriate ioctl for device

of course, both errors are going to stderr.

while trying to generate with request_pty='force' and term_type='ansi',

process = await conn.create_process(request_pty='force', term_type='ansi')
process.stdin.write('puttygen -q -t rsa -b 4096 -o ~/mykey.ppk')

I encounter two problems:

It seems that the key tries to generate, but I can't see it when trying to ls -l ~:

Linux oteto-client-2 5.10.0-21-amd64 #1 SMP Debian 5.10.162-1 (2023-01-21) x86_64

The programs included with the Debian GNU/Linux system are free software;
the exact distribution terms for each program are described in the
individual files in /usr/share/doc/*/copyright.

Debian GNU/Linux comes with ABSOLUTELY NO WARRANTY, to the extent
permitted by applicable law.
Last login: Thu Mar 23 12:49:42 2023 from 10.13.1.123
puttygen -t rsa -b 4096 -o ~/mykey.ppk
qwe
qwe
user@oteto-client-2:~$ puttygen -q -t rsa -b 4096 -o ~/mykey.ppk
Enter passphrase to save key: 
Re-enter passphrase to verify:

I can't handle the clean session closing when using pty. For example, if I need to handle the closing for both non- and interactive modes, calling process.write_eof() working fine for these cases, but not for pty mode. Instead of write_eof() I need to catch an TimeoutError() exception in try-except-else-finally clause and then close the session.

I also want to mention two important things:

I have already implemented .ppk key generation, but with ParallelSSH. It also supports pty, but I assume that the core of that library just works in some another way: it uses greenlets, python threads and just different command(s) execution handling.
I found a clue to use pty in these stackoverflow Q&A's: question and answer, pointed from previous question.

Specifications

Now, let's move to the deep describing of the issues. I will start with some requiered specifications and then move on to my code snippets to demonstate what logic I'm trying to imlement, of course adding some explanations and comments too.

Python version: 3.11.1
AsyncSSH version: 2.13.1, from PyPI
Linux version: let me just execute some commands...

$ uname -a
Linux dev-serv-local 5.10.0-21-amd64 #1 SMP Debian 5.10.162-1 (2023-01-21) x86_64 GNU/Linux

$ cat /etc/os-release
PRETTY_NAME="Debian GNU/Linux 11 (bullseye)"
NAME="Debian GNU/Linux"
VERSION_ID="11"
VERSION="11 (bullseye)"
VERSION_CODENAME=bullseye
ID=debian

$ hostnamectl
Static hostname: dev-serv-local
Icon name: computer-laptop
Chassis: laptop
Operating System: Debian GNU/Linux 11 (bullseye)
Kernel: Linux 5.10.0-21-amd64
Architecture: x86-64

Libs installed: output from pip freeze...

aiofiles==23.1.0
aiogram==3.0.0b7
aiohttp==3.8.4
aiosignal==1.3.1
anyio==3.6.2
async-timeout==4.0.2
asyncpg==0.27.0
asyncssh==2.13.1
attrs==22.2.0
Babel==2.9.1
bcrypt==4.0.1
certifi==2022.12.7
cffi==1.15.1
charset-normalizer==3.1.0
click==8.1.3
cryptography==39.0.2
fastapi==0.95.0
frozenlist==1.3.3
greenlet==2.0.2
gunicorn==20.1.0
h11==0.14.0
h2==4.1.0
hpack==4.0.0
httpcore==0.16.3
httptools==0.5.0
httpx==0.23.3
hyperframe==6.0.1
idna==3.4
inflect==6.0.2
iniconfig==2.0.0
itsdangerous==2.1.2
Jinja2==3.1.2
magic-filter==1.0.9
MarkupSafe==2.1.2
multidict==6.0.4
packaging==23.0
pluggy==1.0.0
psycopg-binary==3.1.8
psycopg-pool==3.1.6
pycparser==2.21
pydantic==1.10.7
pyngrok==5.2.1
python-dotenv==1.0.0
python-multipart==0.0.6
pytz==2022.7.1
PyYAML==6.0
redis==4.5.3
rfc3986==1.5.0
sniffio==1.3.0
sqladmin==0.10.0
SQLAlchemy==2.0.7
starlette==0.26.1
typing_extensions==4.5.0
uvicorn==0.21.1
uvloop==0.17.0
watchfiles==0.18.1
websockets==10.4
WTForms==3.0.1
yarl==1.8.2

Code snippets and main ideas

The main idea is to implement 5 general approaches for handling the next cases:

Command executing as a regular user, that logged through asyncssh.connect();
Executing as logged user with sudo privileges;
Executing as another user, with specifying login and password of such one to the SSHClient.__init__();
Same as 3-d, but also with sudo privileges;
Handling ".id_rsa" keys (generated with ssh-keygen) with any of the logic above.

Now, the code itself.

from typing import Optional, Union

from asyncssh import (
    connect as async_ssh_connect, import_private_key, SSHClientConnection,
    SSHClientConnectionOptions, PermissionDenied, ProtocolError, logger,
    TimeoutError, ChannelOpenError, KeyImportError, KeyEncryptionError
    )

class SSHClient():
    def __init__(
            self,
            host:str,
            username:str,
            password:str,
            port:int=22,
            user_to_exec_as:str=None,
            password_of_user_to_exec_as:str=None,
            private_key:str=None,
            pkey_password:str=None,
            login_timeout:int=5
        ) -> None:

        self.host = host
        self.port = port

        self.__user_to_exec_as = user_to_exec_as or None
        self.__password_of_user_to_exec_as = password_of_user_to_exec_as or None

        self.__private_key = private_key or None
        self.__pkey_password = pkey_password or None

        self.__ssh_client_options = SSHClientConnectionOptions(
            username=username,
            password=password,
            passphrase=pkey_password,
            login_timeout=login_timeout
        )

        logger.set_debug_level(2)

Just the __init__ with fields, some of their purpose is described above.

Next, let's see the _try_to_connect() method.

async def _try_to_connect(self) -> SSHClientConnection:
        if self.__private_key and self.__pkey_password:
            try:
                private_key = import_private_key(
                    self.__private_key, self.__pkey_password)

            except KeyImportError as exc:
                raise exc

            else:
                self.__ssh_client_options.client_keys = private_key

        elif self.__private_key and not self.__pkey_password:
            try:
                private_key = import_private_key(self.__private_key)

            except (KeyImportError, KeyEncryptionError) as exc:
                raise exc

            else:
                self.__ssh_client_options.client_keys = private_key

        try:
            connection = await async_ssh_connect(
                host=self.host,
                port=self.port,
                options=self.__ssh_client_options,
                known_hosts=None
            )

        except (
            ConnectionRefusedError, PermissionDenied, ChannelOpenError,
            ChannelOpenError, ProtocolError
            ) as exc:
            raise exc
        else:
            logger.info(f'conn: {connection._peer_addr}')
            return connection

I assume that this method will try to connect to the host, specified in classes fields, and handle all the possible raises of errors from asyncssh module, and then re-raise them to next method - execute():

    async def execute(
            self,
            command:str,
            pty:bool=False,
            sudo:bool=False,
            timeout:int=300,
            interactive_mode:bool=False,
            custom_input:Optional[Union[tuple[str], str]]=False,
        ) -> Union[
            dict[int, KeyImportError],
            dict[int, KeyEncryptionError],
            dict[int, ConnectionRefusedError],
            dict[int, tuple[ProtocolError, str, str]],
            dict[int, tuple[PermissionDenied, str, str]],
            dict[int, tuple[ChannelOpenError, str, str]],
            dict[int, list[
                tuple[TimeoutError, str, str], tuple[str, str, int]
            ]],
            tuple[str, str, int]
        ]:

        if self.__user_to_exec_as and not self.__password_of_user_to_exec_as:
            return {-1: ValueError(
                f'Arg user_to_exec_as cannot be used without '
                f'password. Please, specify password_of_user_to_exec_as too!'
                )
            }

        try:
            conn = await self._try_to_connect()
        except KeyImportError as exc:
            return {-2: exc}
        except KeyEncryptionError as exc:
            return {-3: exc}
        except ConnectionRefusedError as exc:
            return {-4: exc}
        except PermissionDenied as exc:
            return {-5: (exc, exc.reason, exc.lang)}
        except ChannelOpenError as exc:
            return {-6: (exc, exc.reason, exc.lang)}
        except ProtocolError as exc:
            return {-7: (exc, exc.reason, exc.lang)}

        # Executing command(s) when specified `sudo` argument only.
        if sudo:
            try:
                if interactive_mode:
                    process = await conn.create_process()
                    process.stdin.write('sudo -S bash\n')

                elif pty:
                    process = await conn.create_process(
                        request_pty='force', term_type='ansi')
                    process.stdin.write('sudo -S bash\n')

                else:
                    process = await conn.create_process(command='sudo -S bash')

            except ChannelOpenError as exc:
                return {-6: (exc, exc.reason, exc.lang)}

            process.stdin.write(
                f'{self.__ssh_client_options.password}\n')

            process.stdin.writelines(f'{command}\n')

        # Executing command(s) when specified `user_to_exec_as` and
        # `password_of_user_to_exec_as` args.
        elif self.__user_to_exec_as:
            try:
                if interactive_mode:
                    process = await conn.create_process()
                    process.stdin.write(f'su {self.__user_to_exec_as}\n')

                elif pty:
                    process = await conn.create_process(
                        request_pty='force', term_type='ansi')
                    process.stdin.write(f'su {self.__user_to_exec_as}\n')

                else:
                    process = await conn.create_process(
                        f'su {self.__user_to_exec_as}')

            except ChannelOpenError as exc:
                return {-6: (exc, exc.reason, exc.lang)}

            process.stdin.write(
                f'{self.__ssh_client_options.password}\n')

            process.stdin.writelines(f'{command}\n')

        # Executing command(s) when specified `sudo` and `user_to_exec_as`
        # with `password_of_user_to_exec_as` args.
        elif sudo and self.__user_to_exec_as:
            try:
                if interactive_mode:
                    process = await conn.create_process()
                    process.stdin.write(
                        f'su {self.__user_to_exec_as} -c \'sudo -S bash\'\n')

                elif pty:
                    process = await conn.create_process(
                        request_pty='force', term_type='ansi')
                    process.stdin.write(
                        f'su {self.__user_to_exec_as} -c \'sudo -S bash\'\n')

                else:
                    process = await conn.create_process(
                        f'su {self.__user_to_exec_as} -c \'sudo -S bash\'')

            except ChannelOpenError as exc:
                return {-6: (exc, exc.reason, exc.lang)}

            process.stdin.write(
                f'{self.__ssh_client_options.password}\n')
            process.stdin.write(
                f'{self.__password_of_user_to_exec_as}\n')
            process.stdin.write(f'{command}\n')

        # Executing command(s) just as regular user.
        else:
            try:
                if interactive_mode:
                    process = await conn.create_process()

                elif pty:
                    process = await conn.create_process(term_type='ansi')

                else:
                    process = await conn.create_process('bash')

            except ChannelOpenError as exc:
                return {-6: (exc, exc.reason, exc.lang)}

            process.stdin.write(f'{command}\n')

        # If we have some additional input, such as passwords for some commands.
        if custom_input:
            if isinstance(custom_input, tuple):
                for cmd in custom_input:
                    process.stdin.write(f'{cmd}\n')
            else:
                process.stdin.writelines(f'{custom_input}\n')

        process.stdin.write_eof()

        try:
            completed_process = await process.wait(timeout=timeout)

        except TimeoutError as exc:
            return {-8:
                    [
                        (exc, exc.reason, exc.lang),
                        (exc.stdout, exc.stderr, exc.exit_status)
                    ]
                }
        else:
            stdout:str = completed_process.stdout
            stderr:str = completed_process.stderr
            exit_status:int = completed_process.exit_status
            return stdout, stderr, exit_status
        finally:
            process.close()
            await process.wait_closed()
            conn.close()
            await conn.wait_closed()

This is the main nethod that I'm using to execute command(s) on host and get some output, of course.

Unit-testing and deep problem explanation

Now, let's reproduce the problem itself: we will execute some simple unit-tests that I prepared.

from unittest.async_case import IsolatedAsyncioTestCase
import logging, sys

from src.services.ssh.base import SSHClient, TimeoutError

TIMEOUT = 3
logger = logging.getLogger(__name__)
logger.level = logging.DEBUG

def log(msg: str) -> None:
    stream_handler = logging.StreamHandler(sys.stdout)
    logger.addHandler(stream_handler)

    logger.debug(msg)

    logger.removeHandler(stream_handler)

class TestSSH(IsolatedAsyncioTestCase):

    async def test_execute(self):
        ssh_client = SSHClient(
            host='10.13.1.72',
            username='user',
            password='qwe'
            )

        result = await ssh_client.execute(command='whoami', timeout=TIMEOUT)

        self.assertEqual(result[0], 'user\n') # checking stdout
        self.assertEqual(result[2], 0) # checking exit_status

    async def test_execute_puttygen(self):
        ssh_client = SSHClient(
            host='10.13.1.72',
            username='user',
            password='qwe'
            )

        result_puttygen = await ssh_client.execute(
            command='puttygen -q -t rsa -b 4096 -o ~/mykey.ppk',
            timeout=30,
            custom_input=('qwe', 'qwe')
            )

        log(f'\n{self.id()}\nSTDOUT [puttygen]:\n{result_puttygen[0]}\n\n') # printing stdout
        log(f'\n{self.id()}\nSTDERR [puttygen]:\n{result_puttygen[1]}\n\n') # printing stderr
        self.assertEqual(result_puttygen[2], 1) # checking exit_status

        result_ls = await ssh_client.execute(command='ls -l ~')
        log(f'\n{self.id()}\nSTDOUT [ls -l ~]:\n{result_ls[0]}\n\n') # printing stdout

class TestInteractiveSSH(IsolatedAsyncioTestCase):

    async def test_execute(self):
        ssh_client = SSHClient(
            host='10.13.1.72',
            username='user',
            password='qwe'
            )

        result = await ssh_client.execute(
            command='whoami',
            interactive_mode=True,
            timeout=TIMEOUT
            )

        self.assertEqual(result[2], 0) # checking exit_status

    async def test_execute_puttygen(self):
        ssh_client = SSHClient(
            host='10.13.1.72',
            username='user',
            password='qwe'
            )

        result_puttygen = await ssh_client.execute(
            command='puttygen -q -t rsa -b 4096 -o ~/mykey.ppk',
            interactive_mode=True,
            timeout=30,
            custom_input=('qwe', 'qwe')
            )

        log(f'\n{self.id()}\nSTDOUT [puttygen]:\n{result_puttygen[0]}\n\n') # printing stdout
        log(f'\n{self.id()}\nSTDERR [puttygen]:\n{result_puttygen[1]}\n\n') # printing stderr
        self.assertEqual(result_puttygen[2], 1) # checking exit_status

        result_ls = await ssh_client.execute(command='ls -l ~', interactive_mode=True)
        log(f'\n{self.id()}\nSTDOUT [ls -l ~]:\n{result_ls[0]}\n\n') # printing stdout

class TestPtySSH(IsolatedAsyncioTestCase):

    async def test_execute(self):
        ssh_client = SSHClient(
            host='10.13.1.72',
            username='user',
            password='qwe'
            )

        result = await ssh_client.execute(
            command='whoami',
            pty=True,
            timeout=TIMEOUT
            )

        if isinstance(result, dict):
            self.assertIsInstance(result.get(-8)[0][0], TimeoutError)
            log(f'\n{self.id()}\nSTDOUT:\n{result.get(-8)[1][0]}\n\n') # printing partial stdout
            log(f'\n{self.id()}\nSTDERR:\n{result.get(-8)[1][1]}\n\n') # printing partial stderr

        elif isinstance(result, tuple):
            self.assertEqual(result[2], 0) # checking exit_status

    async def test_execute_with_sudo(self):
        ssh_client = SSHClient(
            host='10.13.1.72',
            username='user',
            password='qwe'
            )

        result = await ssh_client.execute(
            command='whoami',
            pty=True,
            sudo=True,
            timeout=TIMEOUT
            )

        if isinstance(result, dict):
            self.assertIsInstance(result.get(-8)[0][0], TimeoutError)
            log(f'\n{self.id()}\nSTDOUT:\n{result.get(-8)[1][0]}\n\n') # printing partial stdout
            log(f'\n{self.id()}\nSTDERR:\n{result.get(-8)[1][1]}\n\n') # printing partial stderr

        elif isinstance(result, tuple):
            self.assertEqual(result[2], 0) # checking exit_status

    async def test_execute_with_user_to_exec(self):
        ssh_client = SSHClient(
            host='10.13.1.72',
            username='user',
            password='qwe',
            user_to_exec_as='user',
            password_of_user_to_exec_as='qwe'
            )

        result = await ssh_client.execute(
            command='whoami',
            pty=True,
            timeout=TIMEOUT
            )

        if isinstance(result, dict):
            self.assertIsInstance(result.get(-8)[0][0], TimeoutError)
            log(f'\n{self.id()}\nSTDOUT:\n{result.get(-8)[1][0]}\n\n') # printing partial stdout
            log(f'\n{self.id()}\nSTDERR:\n{result.get(-8)[1][1]}\n\n') # printing partial stderr

        elif isinstance(result, tuple):
            self.assertEqual(result[2], 0) # checking exit_status

    async def test_execute_with_user_to_exec_and_sudo(self):
        ssh_client = SSHClient(
            host='10.13.1.72',
            username='user',
            password='qwe',
            user_to_exec_as='user',
            password_of_user_to_exec_as='qwe'
            )

        result = await ssh_client.execute(
            command='whoami',
            pty=True,
            sudo=True,
            timeout=TIMEOUT
            )

        if isinstance(result, dict):
            self.assertIsInstance(result.get(-8)[0][0], TimeoutError)
            log(f'\n{self.id()}\nSTDOUT:\n{result.get(-8)[1][0]}\n\n') # printing partial stdout
            log(f'\n{self.id()}\nSTDERR:\n{result.get(-8)[1][1]}\n\n') # printing partial stderr

        elif isinstance(result, tuple):
            self.assertEqual(result[2], 0) # checking exit_status

    async def test_execute_puttygen(self):
        ssh_client = SSHClient(
            host='10.13.1.72',
            username='user',
            password='qwe'
            )

        result_puttygen = await ssh_client.execute(
            command='puttygen -t rsa -b 4096 -o ~/mykey.ppk',
            pty=True,
            timeout=30,
            custom_input=('qwe', 'qwe')
            )

        if isinstance(result_puttygen, dict):
            self.assertIsInstance(result_puttygen.get(-8)[0][0], TimeoutError)
            log(f'\n{self.id()}\nSTDOUT:\n{result_puttygen.get(-8)[1][0]}\n\n') # printing partial stdout
            log(f'\n{self.id()}\nSTDERR:\n{result_puttygen.get(-8)[1][1]}\n\n') # printing partial stderr

            result_ls = await ssh_client.execute(command='ls -l ~')
            log(f'\n{self.id()}\nSTDOUT [ls -l ~]:\n{result_ls[0]}\n\n') # printing stdout

        elif isinstance(result_puttygen, tuple):
            self.assertEqual(result_puttygen[2], 0) # checking exit_status

            result_ls = await ssh_client.execute(command='ls -l ~')
            log(f'\n{self.id()}\nSTDOUT [ls -l ~]:\n{result_ls[0]}\n\n') # printing stdout

I dropped some of tests to reduce the code filling in the issue, but such tests as test_execute_with_sudo(), test_execute_with_user_to_exec() and test_execute_with_user_to_exec_and_sudo() are correctly passing for TestSSH and TestInteractiveSSH cases.

Now, if we execute all of this bunch of code with python -m unittest tests.test_base_ssh, the next output will arrive:

.
tests.test_base_ssh.TestInteractiveSSH.test_execute_puttygen
STDOUT [puttygen]:
Linux oteto-client-2 5.10.0-21-amd64 #1 SMP Debian 5.10.162-1 (2023-01-21) x86_64

The programs included with the Debian GNU/Linux system are free software;
the exact distribution terms for each program are described in the
individual files in /usr/share/doc/*/copyright.

Debian GNU/Linux comes with ABSOLUTELY NO WARRANTY, to the extent
permitted by applicable law.

tests.test_base_ssh.TestInteractiveSSH.test_execute_puttygen
STDERR [puttygen]:
Enter passphrase to save key: 
Re-enter passphrase to verify: 
puttygen: unable to read new passphrase: Inappropriate ioctl for device

tests.test_base_ssh.TestInteractiveSSH.test_execute_puttygen
STDOUT [ls -l ~]:
Linux oteto-client-2 5.10.0-21-amd64 #1 SMP Debian 5.10.162-1 (2023-01-21) x86_64

The programs included with the Debian GNU/Linux system are free software;
the exact distribution terms for each program are described in the
individual files in /usr/share/doc/*/copyright.

Debian GNU/Linux comes with ABSOLUTELY NO WARRANTY, to the extent
permitted by applicable law.
total 0

.
tests.test_base_ssh.TestPtySSH.test_execute
STDOUT:
Linux oteto-client-2 5.10.0-21-amd64 #1 SMP Debian 5.10.162-1 (2023-01-21) x86_64

The programs included with the Debian GNU/Linux system are free software;
the exact distribution terms for each program are described in the
individual files in /usr/share/doc/*/copyright.

Debian GNU/Linux comes with ABSOLUTELY NO WARRANTY, to the extent
permitted by applicable law.
Last login: Thu Mar 23 15:05:17 2023 from 10.13.1.123
whoami
user@oteto-client-2:~$ whoami
user
user@oteto-client-2:~$ 

tests.test_base_ssh.TestPtySSH.test_execute
STDERR:

.
tests.test_base_ssh.TestPtySSH.test_execute_puttygen
STDOUT:
Linux oteto-client-2 5.10.0-21-amd64 #1 SMP Debian 5.10.162-1 (2023-01-21) x86_64

The programs included with the Debian GNU/Linux system are free software;
the exact distribution terms for each program are described in the
individual files in /usr/share/doc/*/copyright.

Debian GNU/Linux comes with ABSOLUTELY NO WARRANTY, to the extent
permitted by applicable law.
Last login: Thu Mar 23 15:08:05 2023 from 10.13.1.123
puttygen -t rsa -b 4096 -o ~/mykey.ppk
qwe
qwe
user@oteto-client-2:~$ puttygen -t rsa -b 4096 -o ~/mykey.ppk
+++++++++++++++++++++++++
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+++++
Enter passphrase to save key: 
Re-enter passphrase to verify: 

tests.test_base_ssh.TestPtySSH.test_execute_puttygen
STDERR:

tests.test_base_ssh.TestPtySSH.test_execute_puttygen
STDOUT [ls -l ~]:
total 0

.
tests.test_base_ssh.TestPtySSH.test_execute_with_sudo
STDOUT:
Linux oteto-client-2 5.10.0-21-amd64 #1 SMP Debian 5.10.162-1 (2023-01-21) x86_64

The programs included with the Debian GNU/Linux system are free software;
the exact distribution terms for each program are described in the
individual files in /usr/share/doc/*/copyright.

Debian GNU/Linux comes with ABSOLUTELY NO WARRANTY, to the extent
permitted by applicable law.
Last login: Thu Mar 23 15:08:09 2023 from 10.13.1.123
sudo -S bash
qwe
whoami
user@oteto-client-2:~$ sudo -S bash
[sudo] password for user: 
root@oteto-client-2:/home/user# 

tests.test_base_ssh.TestPtySSH.test_execute_with_sudo
STDERR:

.
tests.test_base_ssh.TestPtySSH.test_execute_with_user_to_exec
STDOUT:
Linux oteto-client-2 5.10.0-21-amd64 #1 SMP Debian 5.10.162-1 (2023-01-21) x86_64

The programs included with the Debian GNU/Linux system are free software;
the exact distribution terms for each program are described in the
individual files in /usr/share/doc/*/copyright.

Debian GNU/Linux comes with ABSOLUTELY NO WARRANTY, to the extent
permitted by applicable law.
Last login: Thu Mar 23 15:08:39 2023 from 10.13.1.123
su user
qwe
whoami
user@oteto-client-2:~$ su user
Password: 

tests.test_base_ssh.TestPtySSH.test_execute_with_user_to_exec
STDERR:

.
tests.test_base_ssh.TestPtySSH.test_execute_with_user_to_exec_and_sudo
STDOUT:
Linux oteto-client-2 5.10.0-21-amd64 #1 SMP Debian 5.10.162-1 (2023-01-21) x86_64

The programs included with the Debian GNU/Linux system are free software;
the exact distribution terms for each program are described in the
individual files in /usr/share/doc/*/copyright.

Debian GNU/Linux comes with ABSOLUTELY NO WARRANTY, to the extent
permitted by applicable law.
Last login: Thu Mar 23 15:08:43 2023 from 10.13.1.123
sudo -S bash
qwe
whoami
user@oteto-client-2:~$ sudo -S bash
[sudo] password for user: 
root@oteto-client-2:/home/user# 

tests.test_base_ssh.TestPtySSH.test_execute_with_user_to_exec_and_sudo
STDERR:

..
tests.test_base_ssh.TestSSH.test_execute_puttygen
STDOUT [puttygen]:

tests.test_base_ssh.TestSSH.test_execute_puttygen
STDERR [puttygen]:
Enter passphrase to save key: 
Re-enter passphrase to verify: 
puttygen: unable to read new passphrase: Inappropriate ioctl for device

tests.test_base_ssh.TestSSH.test_execute_puttygen
STDOUT [ls -l ~]:
total 0

.
----------------------------------------------------------------------
Ran 9 tests in 73.455s

OK

So, as we can see, I managed to handle the puttygen command execution in TestPtySSH.test_execute_puttygen() with assertIsInstance(), pointing it to TimeoutError(). This demonstrates that shell itself is never closed only if the specified timeout is not reached in process.wait(timeout). We also can mention that the key itself was not generated (or even just not saved, I assume).

Thoughts

puttygen problem: i have an idea that maybe will solve the .ppk key generation problem. I noticed that the command puttygen takes some time first before prompting me to input the passphrase. This time is different at every execution, and I think it's because of randomized key generation. So, maybe I just need to wait before the program is done and prompts me to enter the data? But I just have no idea how to implement that waiting...
pty infinite session problem: i noticed one interesting line in one of @ronf's tests, but this one is related to handling requests on a server side, while I'm implementing the client-side. Maybe I need to start my own-configured server on the remote host somehow?

Questions

How can I handle the puttygen command execution correctly?
What I have to do with never-closed pty session?
It seems that I don't understand the term_type arg. Watching tests and sources of asyncssh I still could neither understand what types of these are actual can be useful for me, nor what the difference between them.
Optional: if any of the readers have comments, wishes or offers of how all of this code can be impoved, both in readability and perfomance ways - I will very appreciate it!

ronf commented 1 year ago

Hello...

I've got a suggestion for how you might make this work in a pretty simple way, but first let me cover some of your initial questions:

The error "Inappropriate ioctl for device" is definitely because you didn't allocate a pty and it wasn't to be able to blank our the entry of the passphrase. However, an easy workaround for this is to specify "--new-passphrase" as an argument to puttygen.. In fact, you can even specify "--new-passphrase /dev/stdin" and still feed the passphrase to stdin, and that works without having a pty, and also avoids needing to specify the passphrase twice.

Most of the rest of the issues you are running into sound like the kinds of things you'll run into when trying to do interactive I/O with a shell, rather than specifying the command to run when you start up the session. It's hard to know when a command finishes executing in this case, unless you send something like "exit" to the shell after you run puttygen. However, there are easier ways to deal with this, at least in this case.

Basically, the key points are to pass the command to run when calling run() instead of trying to use a shell, and to pass in the passphrase without puttygen prompting for it. Here's an example which runs the command and outputs the "ls" output on the resulting file:

import asyncio, asyncssh

async def run_puttygen(host, key_file, passphrase, opts):
    async with asyncssh.connect(host) as conn:
        result = await conn.run(f'puttygen {opts} -o {key_file} --new-passphrase /dev/stdin; '
                                f'ls -l {key_file}', input=passphrase)

    return result.stdout

print(asyncio.run(run_puttygen('localhost', '~/my_key.ppk', 'qwe',  '-t rsa -b 4096')), end='')

As you can see, you can pass in multiple commands to execute sequentially without resorting to having to send the commands as input to an interactive shell. This also allows you to use the input= to pass in a passphrase, and by telling puttygen to get the passphrase from /dev/stdin, it avoids needing a pty and avoids needing to output the passphrase multiple times or add newlines to it.

You can tweak the set of commands to run if you want to do something other than get the \ls\ output back. Basically, anything you write to stdout will end up being returned by the call to run_puttygen().

Because you are running a command, you don't have to worry about the shell never exiting, but as noted above you could have solved that by sending an explicit "exit" command as well.

As for term_type, it doesn't really matter what you set it to in this case, as you're not running something like a curses application or other program which relies on sending terminal escape sequences. A good generic value is 'ansi'. That said, the only point of setting term_type here would be to avoid needing the request_pty='force'. In my version, you don't need either of these, though.

ronf commented 1 year ago

By the way, if you want output from both stdout and stderr to use in error handling, you could modify run_puttygen() to do something like return result.stdout, result.stderr to get back a tuple of both stdout and stderr output. You could also return the result object, which has a bunch of other information about the result of running the commands. See the SSHCompletedProcess class for details.

lastochka364 commented 1 year ago

Just tested it - works. I haven't seen anything about --new-passphrase arg in puttygen when was googling my problem, just being too focused on pty "issue".

Seems like I need to recall the KISS. Thanks for your time and such quick response.

ronf / asyncssh