mirror of
https://github.com/GNS3/gns3-server
synced 2024-12-27 01:08:33 +00:00
ac508435c3
Arrows UP/DOWN for telnet, Ref. #1083 Typo, Ref. #1083 Support async commands, Ref. #1083 Small refactor, Ref. #1083 Asyncio telnet server - connections support, Ref. #10831 Prompt-toolkit in dependencies, ref. #1083 Few comments, ref. #1083 Direct imports, ref. #1083 Windows size changed support in telnet server, ref. #1139 Fake termios Fake termios - different approach InputStream - copied source from prompt_toolkit
420 lines
14 KiB
Python
420 lines
14 KiB
Python
"""
|
|
Parser for VT100 input stream.
|
|
"""
|
|
|
|
# Copied from prompt_toolkit/terminal/vt100_input.py due to dependency on termios (which is not available on Windows)
|
|
|
|
from __future__ import unicode_literals
|
|
|
|
import re
|
|
import six
|
|
|
|
from six.moves import range
|
|
|
|
from prompt_toolkit.keys import Keys
|
|
from prompt_toolkit.key_binding.input_processor import KeyPress
|
|
|
|
__all__ = (
|
|
'InputStream',
|
|
'raw_mode',
|
|
'cooked_mode',
|
|
)
|
|
|
|
_DEBUG_RENDERER_INPUT = False
|
|
_DEBUG_RENDERER_INPUT_FILENAME = 'prompt-toolkit-render-input.log'
|
|
|
|
|
|
# Regex matching any CPR response
|
|
# (Note that we use '\Z' instead of '$', because '$' could include a trailing
|
|
# newline.)
|
|
_cpr_response_re = re.compile('^' + re.escape('\x1b[') + r'\d+;\d+R\Z')
|
|
|
|
# Mouse events:
|
|
# Typical: "Esc[MaB*" Urxvt: "Esc[96;14;13M" and for Xterm SGR: "Esc[<64;85;12M"
|
|
_mouse_event_re = re.compile('^' + re.escape('\x1b[') + r'(<?[\d;]+[mM]|M...)\Z')
|
|
|
|
# Regex matching any valid prefix of a CPR response.
|
|
# (Note that it doesn't contain the last character, the 'R'. The prefix has to
|
|
# be shorter.)
|
|
_cpr_response_prefix_re = re.compile('^' + re.escape('\x1b[') + r'[\d;]*\Z')
|
|
|
|
_mouse_event_prefix_re = re.compile('^' + re.escape('\x1b[') + r'(<?[\d;]*|M.{0,2})\Z')
|
|
|
|
|
|
class _Flush(object):
|
|
""" Helper object to indicate flush operation to the parser. """
|
|
pass
|
|
|
|
|
|
# Mapping of vt100 escape codes to Keys.
|
|
ANSI_SEQUENCES = {
|
|
'\x1b': Keys.Escape,
|
|
|
|
'\x00': Keys.ControlSpace, # Control-Space (Also for Ctrl-@)
|
|
'\x01': Keys.ControlA, # Control-A (home)
|
|
'\x02': Keys.ControlB, # Control-B (emacs cursor left)
|
|
'\x03': Keys.ControlC, # Control-C (interrupt)
|
|
'\x04': Keys.ControlD, # Control-D (exit)
|
|
'\x05': Keys.ControlE, # Contrel-E (end)
|
|
'\x06': Keys.ControlF, # Control-F (cursor forward)
|
|
'\x07': Keys.ControlG, # Control-G
|
|
'\x08': Keys.ControlH, # Control-H (8) (Identical to '\b')
|
|
'\x09': Keys.ControlI, # Control-I (9) (Identical to '\t')
|
|
'\x0a': Keys.ControlJ, # Control-J (10) (Identical to '\n')
|
|
'\x0b': Keys.ControlK, # Control-K (delete until end of line; vertical tab)
|
|
'\x0c': Keys.ControlL, # Control-L (clear; form feed)
|
|
'\x0d': Keys.ControlM, # Control-M (13) (Identical to '\r')
|
|
'\x0e': Keys.ControlN, # Control-N (14) (history forward)
|
|
'\x0f': Keys.ControlO, # Control-O (15)
|
|
'\x10': Keys.ControlP, # Control-P (16) (history back)
|
|
'\x11': Keys.ControlQ, # Control-Q
|
|
'\x12': Keys.ControlR, # Control-R (18) (reverse search)
|
|
'\x13': Keys.ControlS, # Control-S (19) (forward search)
|
|
'\x14': Keys.ControlT, # Control-T
|
|
'\x15': Keys.ControlU, # Control-U
|
|
'\x16': Keys.ControlV, # Control-V
|
|
'\x17': Keys.ControlW, # Control-W
|
|
'\x18': Keys.ControlX, # Control-X
|
|
'\x19': Keys.ControlY, # Control-Y (25)
|
|
'\x1a': Keys.ControlZ, # Control-Z
|
|
|
|
'\x1c': Keys.ControlBackslash, # Both Control-\ and Ctrl-|
|
|
'\x1d': Keys.ControlSquareClose, # Control-]
|
|
'\x1e': Keys.ControlCircumflex, # Control-^
|
|
'\x1f': Keys.ControlUnderscore, # Control-underscore (Also for Ctrl-hypen.)
|
|
'\x7f': Keys.Backspace, # (127) Backspace
|
|
'\x1b[A': Keys.Up,
|
|
'\x1b[B': Keys.Down,
|
|
'\x1b[C': Keys.Right,
|
|
'\x1b[D': Keys.Left,
|
|
'\x1b[H': Keys.Home,
|
|
'\x1bOH': Keys.Home,
|
|
'\x1b[F': Keys.End,
|
|
'\x1bOF': Keys.End,
|
|
'\x1b[3~': Keys.Delete,
|
|
'\x1b[3;2~': Keys.ShiftDelete, # xterm, gnome-terminal.
|
|
'\x1b[3;5~': Keys.ControlDelete, # xterm, gnome-terminal.
|
|
'\x1b[1~': Keys.Home, # tmux
|
|
'\x1b[4~': Keys.End, # tmux
|
|
'\x1b[5~': Keys.PageUp,
|
|
'\x1b[6~': Keys.PageDown,
|
|
'\x1b[7~': Keys.Home, # xrvt
|
|
'\x1b[8~': Keys.End, # xrvt
|
|
'\x1b[Z': Keys.BackTab, # shift + tab
|
|
'\x1b[2~': Keys.Insert,
|
|
|
|
'\x1bOP': Keys.F1,
|
|
'\x1bOQ': Keys.F2,
|
|
'\x1bOR': Keys.F3,
|
|
'\x1bOS': Keys.F4,
|
|
'\x1b[[A': Keys.F1, # Linux console.
|
|
'\x1b[[B': Keys.F2, # Linux console.
|
|
'\x1b[[C': Keys.F3, # Linux console.
|
|
'\x1b[[D': Keys.F4, # Linux console.
|
|
'\x1b[[E': Keys.F5, # Linux console.
|
|
'\x1b[11~': Keys.F1, # rxvt-unicode
|
|
'\x1b[12~': Keys.F2, # rxvt-unicode
|
|
'\x1b[13~': Keys.F3, # rxvt-unicode
|
|
'\x1b[14~': Keys.F4, # rxvt-unicode
|
|
'\x1b[15~': Keys.F5,
|
|
'\x1b[17~': Keys.F6,
|
|
'\x1b[18~': Keys.F7,
|
|
'\x1b[19~': Keys.F8,
|
|
'\x1b[20~': Keys.F9,
|
|
'\x1b[21~': Keys.F10,
|
|
'\x1b[23~': Keys.F11,
|
|
'\x1b[24~': Keys.F12,
|
|
'\x1b[25~': Keys.F13,
|
|
'\x1b[26~': Keys.F14,
|
|
'\x1b[28~': Keys.F15,
|
|
'\x1b[29~': Keys.F16,
|
|
'\x1b[31~': Keys.F17,
|
|
'\x1b[32~': Keys.F18,
|
|
'\x1b[33~': Keys.F19,
|
|
'\x1b[34~': Keys.F20,
|
|
|
|
# Xterm
|
|
'\x1b[1;2P': Keys.F13,
|
|
'\x1b[1;2Q': Keys.F14,
|
|
# '\x1b[1;2R': Keys.F15, # Conflicts with CPR response.
|
|
'\x1b[1;2S': Keys.F16,
|
|
'\x1b[15;2~': Keys.F17,
|
|
'\x1b[17;2~': Keys.F18,
|
|
'\x1b[18;2~': Keys.F19,
|
|
'\x1b[19;2~': Keys.F20,
|
|
'\x1b[20;2~': Keys.F21,
|
|
'\x1b[21;2~': Keys.F22,
|
|
'\x1b[23;2~': Keys.F23,
|
|
'\x1b[24;2~': Keys.F24,
|
|
|
|
'\x1b[1;5A': Keys.ControlUp, # Cursor Mode
|
|
'\x1b[1;5B': Keys.ControlDown, # Cursor Mode
|
|
'\x1b[1;5C': Keys.ControlRight, # Cursor Mode
|
|
'\x1b[1;5D': Keys.ControlLeft, # Cursor Mode
|
|
|
|
'\x1b[1;2A': Keys.ShiftUp,
|
|
'\x1b[1;2B': Keys.ShiftDown,
|
|
'\x1b[1;2C': Keys.ShiftRight,
|
|
'\x1b[1;2D': Keys.ShiftLeft,
|
|
|
|
# Tmux sends following keystrokes when control+arrow is pressed, but for
|
|
# Emacs ansi-term sends the same sequences for normal arrow keys. Consider
|
|
# it a normal arrow press, because that's more important.
|
|
'\x1bOA': Keys.Up,
|
|
'\x1bOB': Keys.Down,
|
|
'\x1bOC': Keys.Right,
|
|
'\x1bOD': Keys.Left,
|
|
|
|
'\x1b[5A': Keys.ControlUp,
|
|
'\x1b[5B': Keys.ControlDown,
|
|
'\x1b[5C': Keys.ControlRight,
|
|
'\x1b[5D': Keys.ControlLeft,
|
|
|
|
'\x1bOc': Keys.ControlRight, # rxvt
|
|
'\x1bOd': Keys.ControlLeft, # rxvt
|
|
|
|
'\x1b[200~': Keys.BracketedPaste, # Start of bracketed paste.
|
|
|
|
# Meta + arrow keys. Several terminals handle this differently.
|
|
# The following sequences are for xterm and gnome-terminal.
|
|
# (Iterm sends ESC followed by the normal arrow_up/down/left/right
|
|
# sequences, and the OSX Terminal sends ESCb and ESCf for "alt
|
|
# arrow_left" and "alt arrow_right." We don't handle these
|
|
# explicitely, in here, because would could not distinguesh between
|
|
# pressing ESC (to go to Vi navigation mode), followed by just the
|
|
# 'b' or 'f' key. These combinations are handled in
|
|
# the input processor.)
|
|
'\x1b[1;3D': (Keys.Escape, Keys.Left),
|
|
'\x1b[1;3C': (Keys.Escape, Keys.Right),
|
|
'\x1b[1;3A': (Keys.Escape, Keys.Up),
|
|
'\x1b[1;3B': (Keys.Escape, Keys.Down),
|
|
|
|
# Sequences generated by numpad 5. Not sure what it means. (It doesn't
|
|
# appear in 'infocmp'. Just ignore.
|
|
'\x1b[E': Keys.Ignore, # Xterm.
|
|
'\x1b[G': Keys.Ignore, # Linux console.
|
|
}
|
|
|
|
|
|
class _IsPrefixOfLongerMatchCache(dict):
|
|
"""
|
|
Dictiory that maps input sequences to a boolean indicating whether there is
|
|
any key that start with this characters.
|
|
"""
|
|
def __missing__(self, prefix):
|
|
# (hard coded) If this could be a prefix of a CPR response, return
|
|
# True.
|
|
if (_cpr_response_prefix_re.match(prefix) or _mouse_event_prefix_re.match(prefix)):
|
|
result = True
|
|
else:
|
|
# If this could be a prefix of anything else, also return True.
|
|
result = any(v for k, v in ANSI_SEQUENCES.items() if k.startswith(prefix) and k != prefix)
|
|
|
|
self[prefix] = result
|
|
return result
|
|
|
|
|
|
_IS_PREFIX_OF_LONGER_MATCH_CACHE = _IsPrefixOfLongerMatchCache()
|
|
|
|
|
|
class InputStream(object):
|
|
"""
|
|
Parser for VT100 input stream.
|
|
|
|
Feed the data through the `feed` method and the correct callbacks of the
|
|
`input_processor` will be called.
|
|
|
|
::
|
|
|
|
def callback(key):
|
|
pass
|
|
i = InputStream(callback)
|
|
i.feed('data\x01...')
|
|
|
|
:attr input_processor: :class:`~prompt_toolkit.key_binding.InputProcessor` instance.
|
|
"""
|
|
# Lookup table of ANSI escape sequences for a VT100 terminal
|
|
# Hint: in order to know what sequences your terminal writes to stdin, run
|
|
# "od -c" and start typing.
|
|
def __init__(self, feed_key_callback):
|
|
assert callable(feed_key_callback)
|
|
|
|
self.feed_key_callback = feed_key_callback
|
|
self.reset()
|
|
|
|
if _DEBUG_RENDERER_INPUT:
|
|
self.LOG = open(_DEBUG_RENDERER_INPUT_FILENAME, 'ab')
|
|
|
|
def reset(self, request=False):
|
|
self._in_bracketed_paste = False
|
|
self._start_parser()
|
|
|
|
def _start_parser(self):
|
|
"""
|
|
Start the parser coroutine.
|
|
"""
|
|
self._input_parser = self._input_parser_generator()
|
|
self._input_parser.send(None)
|
|
|
|
def _get_match(self, prefix):
|
|
"""
|
|
Return the key that maps to this prefix.
|
|
"""
|
|
# (hard coded) If we match a CPR response, return Keys.CPRResponse.
|
|
# (This one doesn't fit in the ANSI_SEQUENCES, because it contains
|
|
# integer variables.)
|
|
if _cpr_response_re.match(prefix):
|
|
return Keys.CPRResponse
|
|
|
|
elif _mouse_event_re.match(prefix):
|
|
return Keys.Vt100MouseEvent
|
|
|
|
# Otherwise, use the mappings.
|
|
try:
|
|
return ANSI_SEQUENCES[prefix]
|
|
except KeyError:
|
|
return None
|
|
|
|
def _input_parser_generator(self):
|
|
"""
|
|
Coroutine (state machine) for the input parser.
|
|
"""
|
|
prefix = ''
|
|
retry = False
|
|
flush = False
|
|
|
|
while True:
|
|
flush = False
|
|
|
|
if retry:
|
|
retry = False
|
|
else:
|
|
# Get next character.
|
|
c = yield
|
|
|
|
if c == _Flush:
|
|
flush = True
|
|
else:
|
|
prefix += c
|
|
|
|
# If we have some data, check for matches.
|
|
if prefix:
|
|
is_prefix_of_longer_match = _IS_PREFIX_OF_LONGER_MATCH_CACHE[prefix]
|
|
match = self._get_match(prefix)
|
|
|
|
# Exact matches found, call handlers..
|
|
if (flush or not is_prefix_of_longer_match) and match:
|
|
self._call_handler(match, prefix)
|
|
prefix = ''
|
|
|
|
# No exact match found.
|
|
elif (flush or not is_prefix_of_longer_match) and not match:
|
|
found = False
|
|
retry = True
|
|
|
|
# Loop over the input, try the longest match first and
|
|
# shift.
|
|
for i in range(len(prefix), 0, -1):
|
|
match= self._get_match(prefix[:i])
|
|
if match:
|
|
self._call_handler(match, prefix[:i])
|
|
prefix = prefix[i:]
|
|
found = True
|
|
|
|
if not found:
|
|
self._call_handler(prefix[0], prefix[0])
|
|
prefix = prefix[1:]
|
|
|
|
def _call_handler(self, key, insert_text):
|
|
"""
|
|
Callback to handler.
|
|
"""
|
|
if isinstance(key, tuple):
|
|
for k in key:
|
|
self._call_handler(k, insert_text)
|
|
else:
|
|
if key == Keys.BracketedPaste:
|
|
self._in_bracketed_paste = True
|
|
self._paste_buffer = ''
|
|
else:
|
|
self.feed_key_callback(KeyPress(key, insert_text))
|
|
|
|
def feed(self, data):
|
|
"""
|
|
Feed the input stream.
|
|
|
|
:param data: Input string (unicode).
|
|
"""
|
|
assert isinstance(data, six.text_type)
|
|
|
|
if _DEBUG_RENDERER_INPUT:
|
|
self.LOG.write(repr(data).encode('utf-8') + b'\n')
|
|
self.LOG.flush()
|
|
|
|
# Handle bracketed paste. (We bypass the parser that matches all other
|
|
# key presses and keep reading input until we see the end mark.)
|
|
# This is much faster then parsing character by character.
|
|
if self._in_bracketed_paste:
|
|
self._paste_buffer += data
|
|
end_mark = '\x1b[201~'
|
|
|
|
if end_mark in self._paste_buffer:
|
|
end_index = self._paste_buffer.index(end_mark)
|
|
|
|
# Feed content to key bindings.
|
|
paste_content = self._paste_buffer[:end_index]
|
|
self.feed_key_callback(KeyPress(Keys.BracketedPaste, paste_content))
|
|
|
|
# Quit bracketed paste mode and handle remaining input.
|
|
self._in_bracketed_paste = False
|
|
remaining = self._paste_buffer[end_index + len(end_mark):]
|
|
self._paste_buffer = ''
|
|
|
|
self.feed(remaining)
|
|
|
|
# Handle normal input character by character.
|
|
else:
|
|
for i, c in enumerate(data):
|
|
if self._in_bracketed_paste:
|
|
# Quit loop and process from this position when the parser
|
|
# entered bracketed paste.
|
|
self.feed(data[i:])
|
|
break
|
|
else:
|
|
# Replace \r by \n. (Some clients send \r instead of \n
|
|
# when enter is pressed. E.g. telnet and some other
|
|
# terminals.)
|
|
|
|
# XXX: We should remove this in a future version. It *is*
|
|
# now possible to recognise the difference.
|
|
# (We remove ICRNL/INLCR/IGNCR below.)
|
|
# However, this breaks IPython and maybe other applications,
|
|
# because they bind ControlJ (\n) for handling the Enter key.
|
|
|
|
# When this is removed, replace Enter=ControlJ by
|
|
# Enter=ControlM in keys.py.
|
|
if c == '\r':
|
|
c = '\n'
|
|
self._input_parser.send(c)
|
|
|
|
def flush(self):
|
|
"""
|
|
Flush the buffer of the input stream.
|
|
|
|
This will allow us to handle the escape key (or maybe meta) sooner.
|
|
The input received by the escape key is actually the same as the first
|
|
characters of e.g. Arrow-Up, so without knowing what follows the escape
|
|
sequence, we don't know whether escape has been pressed, or whether
|
|
it's something else. This flush function should be called after a
|
|
timeout, and processes everything that's still in the buffer as-is, so
|
|
without assuming any characters will folow.
|
|
"""
|
|
self._input_parser.send(_Flush)
|
|
|
|
def feed_and_flush(self, data):
|
|
"""
|
|
Wrapper around ``feed`` and ``flush``.
|
|
"""
|
|
self.feed(data)
|
|
self.flush()
|