Missing color escape sequences

[Kentzo]

I successfully create a PTY over an SSH client connection to a network device. However, I do not receive the "color" escape sequences via stdout:

async def main():
    async with contextlib.AsyncExitStack() as stack:
        conn = await stack.enter_async_context(
            asyncssh.connect(
                '...',
                username='test',
                password='test',
                preferred_auth=['password'],
                known_hosts=None,
                config=None
            )
        )
        stdin, stdout, stderr = await conn.open_session(
            env={'LC_ALL': 'C', 'TERM': 'xterm-256color'},
            encoding=None,
            request_pty=True,
            term_type='xterm-256color',
            term_size=(140, 45, 1120, 765),
            term_modes={
                asyncssh.PTY_OP_OSPEED: 9600,
                asyncssh.PTY_OP_ISPEED: 9600,
                asyncssh.PTY_VINTR: 3,
                asyncssh.PTY_VQUIT: 255,
                asyncssh.PTY_VERASE: 127,
                asyncssh.PTY_VKILL: 21,
                asyncssh.PTY_VEOF: 4,
                asyncssh.PTY_VEOL: 255,
                asyncssh.PTY_VEOL2: 255,
                asyncssh.PTY_VSTART: 255,
                asyncssh.PTY_VSTOP: 255,
                asyncssh.PTY_VSUSP: 255,
                asyncssh.PTY_VDSUSP: 255,
                asyncssh.PTY_VREPRINT: 18,
                asyncssh.PTY_WERASE: 23,
                asyncssh.PTY_VLNEXT: 255,
                asyncssh.PTY_VSTATUS: 20,
                asyncssh.PTY_VDISCARD: 255,
                asyncssh.PTY_IGNPAR: 0,
                asyncssh.PTY_PARMRK: 0,
                asyncssh.PTY_INPCK: 0,
                asyncssh.PTY_ISTRIP: 0,
                asyncssh.PTY_INLCR: 1,
                asyncssh.PTY_IGNCR: 0,
                asyncssh.PTY_IUCLC: 1,
                asyncssh.PTY_IXON: 0,
                asyncssh.PTY_IXANY: 1,
                asyncssh.PTY_IXOFF: 0,
                asyncssh.PTY_IMAXBEL: 1,
                asyncssh.PTY_IUTF8: 1,
                asyncssh.PTY_ISIG: 1,
                asyncssh.PTY_ICANON: 1,
                asyncssh.PTY_ECHO: 1,
                asyncssh.PTY_ECHOE: 1,
                asyncssh.PTY_ECHOK: 0,
                asyncssh.PTY_ECHONL: 0,
                asyncssh.PTY_NOFLSH: 0,
                asyncssh.PTY_TOSTOP: 0,
                asyncssh.PTY_IEXTEN: 1,
                asyncssh.PTY_ECHOCTL: 1,
                asyncssh.PTY_ECHOKE: 1,
                asyncssh.PTY_PENDIN: 1,
                asyncssh.PTY_OPOST: 1,
                asyncssh.PTY_ONLCR: 1,
                asyncssh.PTY_OCRNL: 0,
                asyncssh.PTY_ONOCR: 0,
                asyncssh.PTY_ONLRET: 0,
                asyncssh.PTY_CS7: 1,
                asyncssh.PTY_CS8: 1,
                asyncssh.PTY_PARENB: 0,
                asyncssh.PTY_PARODD: 0
            }
        )
        banner = await stdout.readuntil(b'> ')
        print(banner) 

Where the session parameters mimic what the OpenSSH client run from the Terminal app sends.

[ronf]

This works for me. I had to change the b'> ' to just b'>', since my prompt didn't include a space after the '>', and fill in appropriate credentials, but that's all.

It should be sufficient here to just specify term_type='xterm-256color'. You shouldn't need to set that in env manually, or do the request_pty=True since that's automatic once term_type is specified. You also shouldn't need to specify term_size or term_modes.

Here's a minimal version of the code that worked for me (though I also got your version above to work with the changes mentioned):

import asyncio, asyncssh

async def main():
    async with asyncssh.connect('localhost') as conn:
        stdin, stdout, stderr = await conn.open_session(
            encoding=None, term_type='xterm-256color')
        banner = await stdout.readuntil(b'> ')
        print(banner)

asyncio.run(main())

Make sure the shell you run includes the '> ' somewhere in its initial output. Otherwise, the readuntil() will never return.

[Kentzo]

I receive output, however parts of the output that are colored in the Terminal app do not have corresponding escape sequences in AsyncSSH.

My current hypothesis is that the SSH server sends some terminal escape sequences which are ignored by my simple client; that then leads the server to assume the terminal as "dumb" and disregard term_type.

With debug logging enabled via:

logging.basicConfig(level=logging.DEBUG)
asyncssh.set_debug_level(2)

(please let me know if verbosity can be further increased)

I see that the server first sends 603 bytes which end with the x1b[9999B\r\x1b[9999B\r\n\x1bZ \x1b[6n escape sequence; then it waits for 10 seconds; finally it prints the prompt without colors.

[ronf]

You may be onto something -- the '\x1bZ' is asking for the VT100 Identification string. It looks like it is expecting back something of the form ESC [ ? 1 ; attributes c -- search for Device Attributes in https://vt100.net/docs/vt100-ug/chapter3.html.

[ronf]

Playing with this on iTerm 2, it looks like sending ESC Z doesn't do anything, but sending ESC [ 0 c does, returning ESC [ ? 6 2 ; 4 c, which according to https://invisible-island.net/xterm/ctlseqs/ctlseqs.html means "VT220 with sixel graphics support". Terminal.app on the Mac returns ESC [ 1 ; 2 c, which means 'VT100 with advanced video option'. A real xterm returns something much more complicated: ESC [ ? 6 3 ; 1 ; 2 ; 4 ; 6 ; 9 ; 1 5 ; 1 6 ; 2 2 ; 2 8 c, which means "VT320 with support for 132 columns, printer, sixel graphics, selective erase, technical characters, locator port, ANSI color, and rectangular editing".

[Kentzo]

Is there some trick in handling escape sequences with AsyncSSH? I'm trying https://asyncssh.readthedocs.io/en/latest/#simple-server to see what OpenSSH client in Terminal app sends in response to these sequences:

async def handle_client(process: asyncssh.SSHServerProcess) -> None:
    process.stdout.write(b'\x1bZ  \x1b[6n')
    print(await process.stdin.readexactly(7))

In debug logs I see

DEBUG:asyncssh:[conn=0, chan=0] Sending 8 data bytes
DEBUG:asyncssh:[conn=0, chan=0] Received 7 data bytes

1. It appears that either the client ignored \x1bZ or something happened in transit, because it was not replied
2. The reply for \x1b[6n (cursor position) was \x1b[45;3R

[Kentzo]

but sending ESC [ 0 c does

I don't think I receive that in the "preflight" banner. Here it's in full:

\r\r\n\r\r\n\r\r\n\r\r\n\r\r\n\r\r\n\r\r\n\r\n\r  MMM      MMM       KKK                          TTTTTTTTTTT      KKK\r\n\r  MMMM    MMMM       KKK                          TTTTTTTTTTT      KKK\r\n\r  MMM MMMM MMM  III  KKK  KKK  RRRRRR     OOOOOO      TTT     III  KKK  KKK\r\n\r  MMM  MM  MMM  III  KKKKK     RRR  RRR  OOO  OOO     TTT     III  KKKKK\r\n\r  MMM      MMM  III  KKK KKK   RRRRRR    OOO  OOO     TTT     III  KKK KKK\r\n\r  MMM      MMM  III  KKK  KKK  RRR  RRR   OOOOOO      TTT     III  KKK  KKK\r\n\r\r\n\r  MikroTik RouterOS 7.10.2 (c) 1999-2023       https://www.mikrotik.com/\r\n\r\r\nPress F1 for help\r\n\r\x1b[9999B\r\x1b[9999B\r\n\x1bZ  \x1b[6n\r\r\r\r[scrapli@gateway] >

Note that after 10 seconds the prompt is sent again. I presume that the server redraws prompt after it receives what it expects.

[ronf]

From AsyncSSH's perspective, all of this is raw data. It doesn't do anything special with any of it, except to convert between bytes and str if encoding is set to something other than None. Since you are setting that to None here, though, all the bytes should be passed through exactly as-is, with only the usual Python rules about backslash sequences in strings applying when you are writing string literals in the code.

I did see that some of the terminals in my testing didn't response at all to ESC Z -- that's an older version of the ESC [ 0 c sequence. So, depending on the terminal you are on, it wouldn't surprise me if it ignored that and sent nothing back in response, moving on to the next escape sequence for the cursor position. You should be able to see that even without getting AsyncSSH (or OpenSSH) involved -- just do an echo -n '\x1bZ' in the shell vs. `echo -n '\x1b[0c'. In the latter case, you'll see some input at the beginning of the next prompt, where you might not see this with ESC Z, if you're not on an actual xterm or something very compatible with it.

[Kentzo]

Looks like it was indeed the exchange of escape sequences that missed server's expectations. I modified the simple server to install a MitM:

async def handle_client(left_process: asyncssh.SSHServerProcess) -> None:
    async with contextlib.AsyncExitStack() as stack:
        await stack.enter_async_context(left_process)

        conn = await stack.enter_async_context(
            asyncssh.connect(
                '192.168.2.1',
                username='test',
                password='test',
                preferred_auth=['password'],
                known_hosts=None,
                config=None
            )
        )
        right_process: asyncssh.SSHClientProcess = await stack.enter_async_context(
            conn.create_process(
                env=left_process.env,
                encoding=None,
                request_pty=True,
                term_type=left_process.term_type,
                term_size=left_process.term_size,
                term_modes=left_process.term_modes,
            )
        )

        handshake = []

        async def redirect_stdin():
            nonlocal handshake
            n = left_process.channel.get_recv_window()
            while not left_process.stdin.at_eof():
                data = await left_process.stdin.read(n)
                handshake.append(('stdin', data))
                right_process.stdin.write(data)

        async def redirect_stdout():
            nonlocal handshake
            n = right_process.channel.get_recv_window()
            while not right_process.stdout.at_eof():
                data = await right_process.stdout.read(n)
                handshake.append(('stdout', data))
                left_process.stdout.write(data)

        redirect: asyncio.TaskGroup = await stack.enter_async_context(asyncio.TaskGroup())
        stack.callback(redirect.create_task(redirect_stdin()).cancel)
        stack.callback(redirect.create_task(redirect_stdout()).cancel)

        await right_process.channel.wait_closed()

(please let me know if there was a better way to achieve this)

The "handshake" turned out to be:

[('stdout', b'\r\r\n\r\r\n\r\r\n\r\r\n\r\r\n\r\r\n'),
 ('stdout', b'\r\r\n\r\n\r  MMM      MMM       KKK                          TTTTTTTTTTT      KKK\r\n'),
 ('stdout',
  b'\r  MMMM    MMMM       KKK                          TTTTTTTTTTT      KKK\r\n\r  MMM MMMM MMM  III  KKK  KKK  RRRRRR     OOOOOO      TTT     III  KKK  KKK\r\n\r  MMM  MM  MMM  III  KKKKK     RRR  RRR  OOO  OOO     TTT     III  KKKKK\r\n\r  MMM      MMM  III  KKK KKK   RRRRRR    OOO  OOO     TTT     III  KKK KKK\r\n\r  MMM      MMM  III  KKK  KKK  RRR  RRR   OOOOOO      TTT     III  KKK  KKK\r\n\r\r\n\r  MikroTik RouterOS 7.10.2 (c) 1999-2023       https://www.mikrotik.com/\r\n'),
 ('stdout', b'\r\r\n'),
 ('stdout', b'Press F1 for help\r\n'),
 ('stdout', b'\r\x1b[9999B\r\x1b[9999B\r\n'),
 ('stdout', b'\x1bZ  \x1b[6n'),
 ('stdin', b'\x1b[45;3R'),
 ('stdout', b'\x1b[4l\x1b[20l\x1b[?7h\x1b[?5l\x1b[?25h\x1b[H\x1b[9999B\x1b[6n'),
 ('stdin', b'\x1b[45;1R'),
 ('stdout', b'\x1b[H\x1b[9999B\x1bD\x1b[9999A\x1b[6n'),
 ('stdin', b'\x1b[1;1R'),
 ('stdout', b'\x1b[H\x1b[9999C\x1b[6n'),
 ('stdin', b'\x1b[1;141R'),
 ('stdout', b'\x1b[H\xc4\x9bH\x1b[6n\r   '),
 ('stdin', b'\x1b[1;3R'),
 ('stdout', b'\x1b[H\x1b[9999C\x1b[6n \x1b[6n \x1b[6n'),
 ('stdin', b'\x1b[1;141R\x1b[1;141R\x1b[2;2R'),
 ('stdout', b'\x0b\x1b[6n'),
 ('stdin', b'\x1b[3;2R'),
 ('stdout', b'\x1b[?47l\x1b[3;5r\x1b[H\x1b[6n\n\n\n\n\n\n\n\x1b[6n\x1b[9999B\x1b[6n\x1b[r\x1b[1;9999r'),
 ('stdin', b'\x1b[1;1R\x1b[5;1R\x1b[45;1R'),
 ('stdout', b'\r\r\r\x1b[9999B\x1b[K[\x1b[m\x1b[36mtest\x1b[m@\x1b[m\x1b[32mgateway\x1b[m] > '),

[ronf]

Looking at the "handshake" here, I don't see any response to the ESC Z. All I see is what look like responses to the get cursor position calls (which there seem to be far too many of!). So, maybe just responding to that would be good enough. That would get tricky to do properly, though, as you'd have to interpret all the incoming ESC sequences and simulate what they would do on a real terminal, as well as handling whatever other non-ESC output the server is trying to display.

[Kentzo]

Thankfully for my application it’s enough to just complete it

[ronf]

Do you end up needing to send all of the data shown here as coming from stdin, or just something like the first such response? I would sort of expect the values here to change based on things like the terminal window size. You also can't really guarantee that the message boundaries will always remain the same, as there's no guarantee that output may be split into multiple messages, or multiple independent writes might arrive in a single message.

[Kentzo]

My observation was that particular device supplies the same banner (including escape sequences). I do set identical window size as was used by the actual terminal. My terminal simulator simply waits for \x1b[6n and writes back pre-recorded responses.

If that fails I will be looking for a terminal emulator.