Re: [asyncssh-users] SSH Server Detecting Hotkey

[Ron Frederick]

On Nov 9, 2019, at 1:17 AM, Augustus Tan <augustu...@gmail.com> wrote:

Hi, I'm kind of new to AsyncSSH, I wonder if there is anyway to detect hotkeys (ctrl-z, ctrl-d)? 

So far, I'm able to detect it if I edit the source code at channel.py

class SSHServerChannel(SSHChannel):

    """SSH server channel"""
.
.
.
.

    def _accept_data(self, data, datatype=None):

        print(data)

I've tried to subclass this SSHServerChannel, but I'm not sure how to pass it over to SSHServerSession self._chan.

Any idea how I can detect those hotkeys?

The SSHChannel class is not meant to be subclassed. Its implementation is internal to AsyncSSH, and _accept_data() in particular is a private function, so you run the risk of your code breaking in future AsyncSSH releases if you attempt to override its implementation.

If you are running an SSH server which allows pseudo-TTYs to be created and the client is requesting one of those, AsyncSSH will enable line editing by default on stdin. In that case, you can use the register_key() method on the SSH server channel object to set a handler to call when certain keys are pressed. However, that’s mainly meant to let you define your own custom key sequences to modify the input line, so I’m not sure that’ll give you the flexibility to do what you want.

There’s actually already a handler registered in this case for Ctrl-D which would trigger a soft “EOF” to be reported on the channel when you are reading it (where read returns an empty string), so supporting Ctrl-D wouldn’t necessarily require you to register that as a hot key. There’s also already a handler registered for Ctrl-C which would cause the BreakReceived exception to be raised on reads. You might be able to register your own handler for Ctrl-Z using register_key, but it would hard to leverage any of the high-level stream or process API functions if you did this. You’d instead be forced back to using the low-level callback APIs. For instance:

class MySSHServerSession(asyncssh.SSHServerSession):

    def shell_requested(self):

        return True

    def connection_made(self, chan):

        self._chan = chan

    def session_started(self):

        self._chan.register_key('\x1a', self.ctrlz_received)

    def data_received(self, data, datatype):

        print('Got data:', repr(data))

    def soft_eof_received(self):

        print('Got EOF')

    def break_received(self, msec):

        print('Got break')

    def ctrlz_received(self, line, pos):

        print('Got Ctrl-Z')

        return line, pos

class MySSHServer(asyncssh.SSHServer):

    def session_requested(self):

        return MySSHServerSession()

async def start_server():

    await asyncssh.create_server(MySSHServer, '', 8022,

                                 server_host_keys=['ssh_host_key'],

                                 authorized_client_keys='ssh_user_ca')

I could potentially extend the line editor register_key() functionality to allow signals to be raised on the channel, and that would let you use the higher-level API to raise a SignalReceived exception when a certain key was hit, breaking you out of whatever read you were blocked on. This would require a small change to the editor code, but with it you’d end up being able to write something like:

def handle_ctrlz(line, pos):

    return 'CTRLZ', -1

async def handle_client(process):

    process.channel.register_key('\x1a', handle_ctrlz)

    while True:

        try:

            line = await process.stdin.readline()

            if line:

                print('Got input:', repr(line))

            else:

                print('Got EOF')

        except asyncssh.BreakReceived:

            print('Got break')

        except asyncssh.SignalReceived as exc:

            print('Got signal:', exc.signal)

    process.exit(0)

async def start_server():

    await asyncssh.listen('', 8022, server_host_keys=['ssh_host_key'],

                          authorized_client_keys='ssh_user_ca',

                          process_factory=handle_client)

Would this be useful to you?

-- 

Ron Frederick
ro...@timeheart.net

[Augustus Tan]

Yup this is great! Thanks a lot!

[Ron Frederick]

Ok - I just checked support into the “develop” branch for the second block of code shown below, where the SignalReceived exception will be delivered when a key handler returns a negative cursor position. If you have a chance to test it and run into any issues, please let me know!

[Augustus]

Hi, I've written a test code to try out the 'develop' branch. Here's my code:

import asyncio

import asyncssh

class TestSSHSession(asyncssh.SSHServerSession):

    def __init__(self):

        self._input = ''

        self._total = 0

        self._chan = None

    def connection_made(self, chan):

        self._chan = chan

    def shell_requested(self):

        return True

    def key_ctrlz(self, line, pos):

        return 'CtrlZ', -1

    def signal_received(self, signal):

        if signal == 'CtrlZ':

            self.soft_eof_received()

    def session_started(self):

        self._chan.register_key('\x1a', self.key_ctrlz)

        self._chan.write('Testing Ctrl-Z...\n')

    def data_received(self, data, datatype):

        self._input += data

        lines = self._input.split('\n')

        for line in lines[:-1]:

            self._chan.write(line + '\n')

        self._input = lines[-1]

    def eof_received(self):

        self._chan.exit(0)

    def break_received(self, msec):

        self.eof_received()

    def soft_eof_received(self):

        print('Soft Exit')

        self._chan.exit(0)

class TestSSHServer(asyncssh.SSHServer):

    def begin_auth(self, username):

        return False

    def session_requested(self):

        return TestSSHSession()

async def start_server():

    await asyncssh.create_server(TestSSHServer, '', 8022,

                                 server_host_keys=['ssh_host_key'],)

loop = asyncio.get_event_loop()

try:

    loop.run_until_complete(start_server())

    loop.run_forever()

except (OSError, asyncssh.Error, KeyboardInterrupt) as e:

    exit()

It works extremely well for me with no errors, when I clicked on Ctrl-Z button. 

Previously, my key_ctrlz function will call the self.soft_eof_received function, and had to return something.

def key_ctrlz(self, line, pos):

    self.soft_eof_received()

    return line, pos

It will raise an error of 'SSH Connection Error: Channel not open for sending'. I decide to ignore it.

Thanks for helping out!!

[Ron Frederick]

Glad to hear you got this working.

The error you mentioned before the change I made today makes sense. After returning line & pos in the original Ctrl-Z handler, it would try to redraw the input line, but that will fail when this is attempted if your soft_eof_received() function called exit() on the channel already. With the change I made, the line is no longer redrawn when the callback returns a negative position.

This change will be included in the next release. Thanks for helping to test this!