Key monitoring

[RonDingDing]

Let's suppose the programmer want to know every keyboard key press at the client, is there a simple way?

[ronf]

The first thing you'll need to do is to request a PTY on the SSH session where you want this. This can be done by setting the "term_type" argument on the session where you want this. Doing so will disable local echo and buffering and local editing on the client. Once you've got that, you can call something like process.stdin.read(1) to have it return input a byte at a time. This won't correspond directly to "key presses", though, as some keys send a sequence of bytes. Since different keys can start off with the same subset of bytes, you need to keep reading until you have enough to distinguish one key from another before you act on it. The built-in line editor in AsyncSSH has the ability to do all of this, and you can even register for it to call arbitrary code of yours when specific key sequences are received, but you need to tell it exactly which sequences to look for so that it knows which ones it has to distinguish between.

Do you really need to be called on ALL key presses, or just specific ones? Do you only need to be able to match keys which send a single byte, or would you need to handle things like arrow keys or function keys where the key could send multiple bytes?

[RonDingDing]

Basically I am implementing a bash-like interactive ssh console, where the client only ssh to the (asyncssh) server using ssh xxx@xxx.xxx.xxx and start typing. It must have the autocompletion and history functionality like when you press <tab> or <up> and <down> in a linux bash.

[RonDingDing]

And could you please write some basic code for me to follow? Real thanks!

[RonDingDing]

request_pty (bool, 'force', or 'auto') – (optional) Whether or not to request a pseudo-terminal (PTY) for this session. This defaults to True, which means to request a PTY whenever the term_type is set. Other possible values include False to never request a PTY, 'force' to always request a PTY even without term_type being set, or 'auto' to request a TTY when term_type is set but only when starting an interactive shell.

Do you mean this ?

[ronf]

Yes - this is one of the arguments I had in mind, but in this case you are implementing an SSH server, so requesting a PTY will be done by the SSH client. As long as TERM is set in the shell you are running the SSH client from, it should ask for a PTY, and by default AsyncSSH will allow this when running as a server, causing the client to let the server do the echoing and line editing. By default, AsyncSSH will provide a line editor with history and basic readline-like editing without needing to get a shell involved, but it doesn't have any kind of autocompletion capabilities by default. If you have an application that can parse the input line and know what possible completions are, you can use the function register_key() in SSHLineEditorChannel to trigger a callback you provide when the user hits a specific key (like tab, for instance). The details of the autocompletion are application-specific, so it's hard for me to provide a self-contained example which does anything interesting, but it could look something like:

import asyncio, asyncssh, sys

def autocomplete(line, pos):
    line = line[:pos] + 'complete' + line[pos:]
    pos += 8

    return line, pos

async def handle_client(process):
    process.stdout.write('Welcome to my SSH server, %s!\r\n\r\n' %
                         process.get_extra_info('username'))

    process.channel.register_key('\t', autocomplete)

    while True:
        process.stdout.write('prompt>')

        line = await process.stdin.readline()
        if not line:
            break

        process.stdout.write(f'Got command: {line}')

    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)

loop = asyncio.get_event_loop()

try:
    loop.run_until_complete(start_server())
except (OSError, asyncssh.Error) as exc:
    sys.exit('Error starting server: ' + str(exc))

loop.run_forever()

This is a modified version of the AsyncSSH server example which enables autocompletion of static text "complete" whenever tab is hit.

Take a look at the autocomplete() callback, which is called with the current input line and cursor position within that line every time the user hits tab. This function can return a new input line and a new cursor position with the results of performing the auto-completion, looking at the input so far at the specified cursor position to decide what completion (if any) makes sense. See the register_key() documentation for other possible return values. For instance, you can return False to leave the input unchanged but cause the editor to ring the terminal bell when there is no valid completion.

[RonDingDing]

Real thanks Ron! Your reply has solved my problem!