# Copyright (c) 2016-2022 by Ron Frederick <ronf@timeheart.net> and others.
#
# This program and the accompanying materials are made available under
# the terms of the Eclipse Public License v2.0 which accompanies this
# distribution and is available at:
#
# http://www.eclipse.org/legal/epl-2.0/
#
# This program may also be made available under the following secondary
# licenses when the conditions for such availability set forth in the
# Eclipse Public License v2.0 are satisfied:
#
# GNU General Public License, Version 2.0, or any later versions of
# that license
#
# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
#
# Contributors:
# Ron Frederick - initial implementation, API, and documentation
"""Input line editor"""
import re
from functools import partial
from typing import TYPE_CHECKING, Callable, Dict, List
from typing import Optional, Set, Tuple, Union, cast
from unicodedata import east_asian_width
from .session import DataType
if TYPE_CHECKING:
# pylint: disable=cyclic-import
from .channel import SSHServerChannel
from .session import SSHServerSession
_CharDict = Dict[str, object]
_CharHandler = Callable[['SSHLineEditor'], None]
_KeyHandler = Callable[[str, int], Union[bool, Tuple[str, int]]]
_DEFAULT_WIDTH = 80
_ansi_terminals = ('ansi', 'cygwin', 'linux', 'putty', 'screen', 'teraterm',
'cit80', 'vt100', 'vt102', 'vt220', 'vt320', 'xterm',
'xterm-color', 'xterm-16color', 'xterm-256color', 'rxvt',
'rxvt-color')
def _is_wide(ch: str) -> bool:
"""Return display width of character"""
return east_asian_width(ch) in 'WF'
class SSHLineEditor:
"""Input line editor"""
def __init__(self, chan: 'SSHServerChannel[str]',
session: 'SSHServerSession[str]', line_echo: bool,
history_size: int, max_line_length: int, term_type: str,
width: int):
self._chan = chan
self._session = session
self._line_echo = line_echo
self._line_pending = False
self._history_size = history_size if history_size > 0 else 0
self._max_line_length = max_line_length
self._wrap = term_type in _ansi_terminals
self._width = width or _DEFAULT_WIDTH
self._line_mode = True
self._echo = True
self._start_column = 0
self._end_column = 0
self._cursor = 0
self._left_pos = 0
self._right_pos = 0
self._pos = 0
self._line = ''
self._bell_rung = False
self._early_wrap: Set[int] = set()
self._outbuf: List[str] = []
self._keymap: _CharDict = {}
self._key_state = self._keymap
self._erased = ''
self._history: List[str] = []
self._history_index = 0
for func, keys in self._keylist:
for key in keys:
self._add_key(key, func)
self._build_printable()
def _add_key(self, key: str, func: _CharHandler) -> None:
"""Add a key to the keymap"""
keymap = self._keymap
for ch in key[:-1]:
if ch not in keymap:
keymap[ch] = {}
keymap = cast(_CharDict, keymap[ch])
keymap[key[-1]] = func
def _del_key(self, key: str) -> None:
"""Delete a key from the keymap"""
keymap = self._keymap
for ch in key[:-1]:
if ch not in keymap:
return
keymap = cast(_CharDict, keymap[ch])
keymap.pop(key[-1], None)
def _build_printable(self) -> None:
"""Build a regex of printable ASCII non-registered keys"""
def _escape(c: int) -> str:
"""Backslash escape special characters in regex character range"""
ch = chr(c)
return ('\\' if (ch in '-&|[]\\^~') else '') + ch
def _is_printable(ch: str) -> bool:
"""Return if character is printable and has no handler"""
return ch.isprintable() and ch not in keys
pat: List[str] = []
keys = self._keymap.keys()
start = ord(' ')
limit = 0x10000
while start < limit:
while start < limit and not _is_printable(chr(start)):
start += 1
end = start
while _is_printable(chr(end)):
end += 1
pat.append(_escape(start))
if start != end - 1:
pat.append('-' + _escape(end - 1))
start = end + 1
self._printable = re.compile('[' + ''.join(pat) + ']*')
def _char_width(self, pos: int) -> int:
"""Return width of character at specified position"""
return 1 + _is_wide(self._line[pos]) + ((pos + 1) in self._early_wrap)
def _determine_column(self, data: str, column: int,
pos: Optional[int] = None) -> Tuple[str, int]:
"""Determine new output column after output occurs"""
escaped = False
offset = pos
last_wrap_pos = pos
wrapped_data = []
for ch in data:
if ch == '\b':
column -= 1
elif ch == '\x1b':
escaped = True
elif escaped:
if ch == 'm':
escaped = False
else:
if _is_wide(ch) and (column % self._width) == self._width - 1:
column += 1
if pos is not None:
assert last_wrap_pos is not None
assert offset is not None
wrapped_data.append(data[last_wrap_pos - offset:
pos - offset])
last_wrap_pos = pos
self._early_wrap.add(pos)
else:
if pos is not None:
self._early_wrap.discard(pos)
column += 1 + _is_wide(ch)
if pos is not None:
pos += 1
if pos is not None:
assert last_wrap_pos is not None
assert offset is not None
wrapped_data.append(data[last_wrap_pos - offset:])
return ' '.join(wrapped_data), column
else:
return data, column
def _output(self, data: str, pos: Optional[int] = None) -> None:
"""Generate output and calculate new output column"""
idx = data.rfind('\n')
if idx >= 0:
self._outbuf.append(data[:idx+1])
tail = data[idx+1:]
self._cursor = 0
else:
tail = data
data, self._cursor = self._determine_column(tail, self._cursor, pos)
self._outbuf.append(data)
if self._cursor and self._cursor % self._width == 0:
self._outbuf.append(' \b')
def _ring_bell(self) -> None:
"""Ring the terminal bell"""
if not self._bell_rung:
self._outbuf.append('\a')
self._bell_rung = True
def _update_input_window(self, new_pos: int) -> int:
"""Update visible input window when not wrapping onto multiple lines"""
line_len = len(self._line)
if new_pos < self._left_pos:
self._left_pos = new_pos
else:
if new_pos < line_len:
new_pos += 1
pos = self._pos
column = self._cursor
while pos < new_pos:
column += self._char_width(pos)
pos += 1
if column >= self._width:
while column >= self._width:
column -= self._char_width(self._left_pos)
self._left_pos += 1
else:
while self._left_pos > 0:
column += self._char_width(self._left_pos)
if column < self._width:
self._left_pos -= 1
else:
break
column = self._start_column
self._right_pos = self._left_pos
while self._right_pos < line_len:
ch_width = self._char_width(self._right_pos)
if column + ch_width < self._width:
self._right_pos += 1
column += ch_width
else:
break
return column
def _move_cursor(self, column: int) -> None:
"""Move the cursor to selected position in input line"""
start_row = self._cursor // self._width
start_col = self._cursor % self._width
end_row = column // self._width
end_col = column % self._width
if end_row < start_row:
self._outbuf.append('\x1b[' + str(start_row-end_row) + 'A')
elif end_row > start_row:
self._outbuf.append('\x1b[' + str(end_row-start_row) + 'B')
if end_col > start_col:
self._outbuf.append('\x1b[' + str(end_col-start_col) + 'C')
elif end_col < start_col:
self._outbuf.append('\x1b[' + str(start_col-end_col) + 'D')
self._cursor = column
def _move_back(self, column: int) -> None:
"""Move the cursor backward to selected position in input line"""
if self._wrap:
self._move_cursor(column)
else:
self._outbuf.append('\b' * (self._cursor - column))
self._cursor = column
def _clear_to_end(self) -> None:
"""Clear any remaining characters from previous input line"""
column = self._cursor
remaining = self._end_column - column
if remaining > 0:
self._outbuf.append(' ' * remaining)
self._cursor = self._end_column
if self._cursor % self._width == 0:
self._outbuf.append(' \b')
self._move_back(column)
self._end_column = column
def _erase_input(self) -> None:
"""Erase current input line"""
self._move_cursor(self._start_column)
self._clear_to_end()
self._early_wrap.clear()
def _draw_input(self) -> None:
"""Draw current input line"""
if self._line and self._echo:
if self._wrap:
self._output(self._line[:self._pos], 0)
column = self._cursor
self._output(self._line[self._pos:], self._pos)
else:
self._update_input_window(self._pos)
self._output(self._line[self._left_pos:self._pos])
column = self._cursor
self._output(self._line[self._pos:self._right_pos])
self._end_column = self._cursor
self._move_back(column)
def _reposition(self, new_pos: int, new_column: int) -> None:
"""Reposition the cursor to selected position in input"""
if self._echo:
if self._wrap:
self._move_cursor(new_column)
else:
self._update_input(self._pos, self._cursor, new_pos)
self._pos = new_pos
def _update_input(self, pos: int, column: int, new_pos: int) -> None:
"""Update selected portion of current input line"""
if self._echo:
if self._wrap:
if pos in self._early_wrap:
column -= 1
self._move_cursor(column)
prev_wrap = new_pos in self._early_wrap
self._output(self._line[pos:new_pos], pos)
column = self._cursor
self._output(self._line[new_pos:], new_pos)
column += (new_pos in self._early_wrap) - prev_wrap
else:
self._update_input_window(new_pos)
self._move_back(self._start_column)
self._output(self._line[self._left_pos:new_pos])
column = self._cursor
self._output(self._line[new_pos:self._right_pos])
self._clear_to_end()
self._move_back(column)
self._pos = new_pos
def _reset_line(self) -> None:
"""Reset input line to empty"""
self._line = ''
self._left_pos = 0
self._right_pos = 0
self._pos = 0
self._start_column = self._cursor
self._end_column = self._cursor
def _reset_pending(self) -> None:
"""Reset a pending echoed line if any"""
if self._line_pending:
self._erase_input()
self._reset_line()
self._line_pending = False
def _insert_printable(self, data: str) -> None:
"""Insert data into the input line"""
line_len = len(self._line)
data_len = len(data)
if self._max_line_length:
if line_len + data_len > self._max_line_length:
self._ring_bell()
data_len = self._max_line_length - line_len
data = data[:data_len]
if data:
pos = self._pos
new_pos = pos + data_len
self._line = self._line[:pos] + data + self._line[pos:]
self._update_input(pos, self._cursor, new_pos)
def _end_line(self) -> None:
"""End the current input line and send it to the session"""
line = self._line
need_wrap = (self._echo and not self._wrap and
(self._left_pos > 0 or self._right_pos < len(line)))
if self._line_echo or need_wrap:
if need_wrap:
self._output('\b' * (self._cursor - self._start_column) + line)
else:
self._move_to_end()
self._output('\r\n')
self._reset_line()
else:
self._move_to_end()
self._line_pending = True
if self._echo and self._history_size and line:
self._history.append(line)
self._history = self._history[-self._history_size:]
self._history_index = len(self._history)
self._session.data_received(line + '\n', None)
def _eof_or_delete(self) -> None:
"""Erase character to the right, or send EOF if input line is empty"""
if not self._line:
self._session.soft_eof_received()
else:
self._erase_right()
def _erase_left(self) -> None:
"""Erase character to the left"""
if self._pos > 0:
pos = self._pos - 1
column = self._cursor - self._char_width(pos)
self._line = self._line[:pos] + self._line[pos+1:]
self._update_input(pos, column, pos)
else:
self._ring_bell()
def _erase_right(self) -> None:
"""Erase character to the right"""
if self._pos < len(self._line):
pos = self._pos
self._line = self._line[:pos] + self._line[pos+1:]
self._update_input(pos, self._cursor, pos)
else:
self._ring_bell()
def _erase_line(self) -> None:
"""Erase entire input line"""
self._erased = self._line
self._line = ''
self._update_input(0, self._start_column, 0)
def _erase_to_end(self) -> None:
"""Erase to end of input line"""
pos = self._pos
self._erased = self._line[pos:]
self._line = self._line[:pos]
self._update_input(pos, self._cursor, pos)
def _handle_key(self, key: str, handler: _KeyHandler) -> None:
"""Call an external key handler"""
result = handler(self._line, self._pos)
if result is True:
if key.isprintable():
self._insert_printable(key)
else:
self._ring_bell()
elif result is False:
self._ring_bell()
else:
line, new_pos = cast(Tuple[str, int], result)
if new_pos < 0:
self._session.signal_received(line)
else:
self._line = line
self._update_input(0, self._start_column, new_pos)
def _history_prev(self) -> None:
"""Replace input with previous line in history"""
if self._history_index > 0:
self._history_index -= 1
self._line = self._history[self._history_index]
self._update_input(0, self._start_column, len(self._line))
else:
self._ring_bell()
def _history_next(self) -> None:
"""Replace input with next line in history"""
if self._history_index < len(self._history):
self._history_index += 1
if self._history_index < len(self._history):
self._line = self._history[self._history_index]
else:
self._line = ''
self._update_input(0, self._start_column, len(self._line))
else:
self._ring_bell()
def _move_left(self) -> None:
"""Move left in input line"""
if self._pos > 0:
pos = self._pos - 1
column = self._cursor - self._char_width(pos)
self._reposition(pos, column)
else:
self._ring_bell()
def _move_right(self) -> None:
"""Move right in input line"""
if self._pos < len(self._line):
pos = self._pos
column = self._cursor + self._char_width(pos)
self._reposition(pos + 1, column)
else:
self._ring_bell()
def _move_to_start(self) -> None:
"""Move to start of input line"""
self._reposition(0, self._start_column)
def _move_to_end(self) -> None:
"""Move to end of input line"""
self._reposition(len(self._line), self._end_column)
def _redraw(self) -> None:
"""Redraw input line"""
self._erase_input()
self._draw_input()
def _insert_erased(self) -> None:
"""Insert previously erased input"""
self._insert_printable(self._erased)
def _send_break(self) -> None:
"""Send break to session"""
self._session.break_received(0)
_keylist = ((_end_line, ('\n', '\r', '\x1bOM')),
(_eof_or_delete, ('\x04',)),
(_erase_left, ('\x08', '\x7f')),
(_erase_right, ('\x1b[3~',)),
(_erase_line, ('\x15',)),
(_erase_to_end, ('\x0b',)),
(_history_prev, ('\x10', '\x1b[A', '\x1bOA')),
(_history_next, ('\x0e', '\x1b[B', '\x1bOB')),
(_move_left, ('\x02', '\x1b[D', '\x1bOD')),
(_move_right, ('\x06', '\x1b[C', '\x1bOC')),
(_move_to_start, ('\x01', '\x1b[H', '\x1b[1~')),
(_move_to_end, ('\x05', '\x1b[F', '\x1b[4~')),
(_redraw, ('\x12',)),
(_insert_erased, ('\x19',)),
(_send_break, ('\x03', '\x1b[33~')))
def register_key(self, key: str, handler: _KeyHandler) -> None:
"""Register a handler to be called when a key is pressed"""
self._add_key(key, partial(SSHLineEditor._handle_key,
key=key, handler=handler))
self._build_printable()
def unregister_key(self, key: str) -> None:
"""Remove the handler associated with a key"""
self._del_key(key)
self._build_printable()
def set_input(self, line: str, pos: int) -> None:
"""Set input line and cursor position"""
self._reset_pending()
self._line = line
self._update_input(0, self._start_column, pos)
def set_line_mode(self, line_mode: bool) -> None:
"""Enable/disable input line editing"""
self._reset_pending()
if self._line and not line_mode:
data = self._line
self._erase_input()
self._line = ''
self._session.data_received(data, None)
self._line_mode = line_mode
def set_echo(self, echo: bool) -> None:
"""Enable/disable echoing of input in line mode"""
self._reset_pending()
if self._echo and not echo:
self._erase_input()
self._echo = False
elif echo and not self._echo:
self._echo = True
self._draw_input()
def set_width(self, width: int) -> None:
"""Set terminal line width"""
self._reset_pending()
self._width = width or _DEFAULT_WIDTH
if self._wrap:
_, self._cursor = self._determine_column(self._line,
self._start_column, 0)
self._redraw()
def process_input(self, data: str, datatype: DataType) -> None:
"""Process input from channel"""
if self._line_mode:
data_len = len(data)
idx = 0
while idx < data_len:
self._reset_pending()
ch = data[idx]
idx += 1
if ch in self._key_state:
key_state = self._key_state[ch]
if callable(key_state):
try:
cast(_CharHandler, key_state)(self)
finally:
self._key_state = self._keymap
else:
self._key_state = cast(_CharDict, key_state)
elif self._key_state == self._keymap and ch.isprintable():
match = self._printable.match(data, idx - 1)
assert match is not None
match = match[0]
if match:
self._insert_printable(match)
idx += len(match) - 1
else:
self._insert_printable(ch)
else:
self._key_state = self._keymap
self._ring_bell()
self._bell_rung = False
if self._outbuf:
self._chan.write(''.join(self._outbuf))
self._outbuf.clear()
else:
self._session.data_received(data, datatype)
def process_output(self, data: str) -> None:
"""Process output to channel"""
if self._line_pending:
if data.startswith(self._line):
self._start_column = self._cursor
data = data[len(self._line):]
else:
self._erase_input()
self._reset_line()
self._line_pending = False
data = data.replace('\n', '\r\n')
self._erase_input()
self._output(data)
if not self._wrap:
self._cursor %= self._width
self._start_column = self._cursor
self._end_column = self._cursor
self._draw_input()
self._chan.write(''.join(self._outbuf))
self._outbuf.clear()
[docs]
class SSHLineEditorChannel:
"""Input line editor channel wrapper
When creating server channels with `line_editor` set to `True`,
this class is wrapped around the channel, providing the caller with
the ability to enable and disable input line editing and echoing.
.. note:: Line editing is only available when a pseudo-terminal
is requested on the server channel and the character
encoding on the channel is not set to `None`.
"""
def __init__(self, orig_chan: 'SSHServerChannel[str]',
orig_session: 'SSHServerSession[str]', line_echo: bool,
history_size: int, max_line_length: int):
self._orig_chan = orig_chan
self._orig_session = orig_session
self._line_echo = line_echo
self._history_size = history_size
self._max_line_length = max_line_length
self._editor: Optional[SSHLineEditor] = None
def __getattr__(self, attr: str):
"""Delegate most channel functions to original channel"""
return getattr(self._orig_chan, attr)
def create_editor(self) -> Optional[SSHLineEditor]:
"""Create input line editor if encoding and terminal type are set"""
encoding, _ = self._orig_chan.get_encoding()
term_type = self._orig_chan.get_terminal_type()
width = self._orig_chan.get_terminal_size()[0]
if encoding and term_type:
self._editor = SSHLineEditor(
self._orig_chan, self._orig_session, self._line_echo,
self._history_size, self._max_line_length, term_type, width)
return self._editor
[docs]
def register_key(self, key: str, handler: _KeyHandler) -> None:
"""Register a handler to be called when a key is pressed
This method registers a handler function which will be called
when a user presses the specified key while inputting a line.
The handler will be called with arguments of the current
input line and cursor position, and updated versions of these
two values should be returned as a tuple.
The handler can also return a tuple of a signal name and
negative cursor position to cause a signal to be delivered
on the channel. In this case, the current input line is left
unchanged but the signal is delivered before processing any
additional input. This can be used to define "hot keys" that
trigger actions unrelated to editing the input.
If the registered key is printable text, returning `True` will
insert that text at the current cursor position, acting as if
no handler was registered for that key. This is useful if you
want to perform a special action in some cases but not others,
such as based on the current cursor position.
Returning `False` will ring the bell and leave the input
unchanged, indicating the requested action could not be
performed.
:param key:
The key sequence to look for
:param handler:
The handler function to call when the key is pressed
:type key: `str`
:type handler: `callable`
"""
assert self._editor is not None
self._editor.register_key(key, handler)
[docs]
def unregister_key(self, key: str) -> None:
"""Remove the handler associated with a key
This method removes a handler function associated with
the specified key. If the key sequence is printable,
this will cause it to return to being inserted at the
current position when pressed. Otherwise, it will cause
the bell to ring to signal the key is not understood.
:param key:
The key sequence to look for
:type key: `str`
"""
assert self._editor is not None
self._editor.unregister_key(key)
def clear_input(self) -> None:
"""Clear input line
This method clears the current input line.
"""
assert self._editor is not None
self._editor.set_input('', 0)
def set_input(self, line: str, pos: int) -> None:
"""Clear input line
This method sets the current input line and cursor position.
:param line:
The new input line
:param pos:
The new cursor position within the input line
:type line: `str`
:type pos: `int`
"""
assert self._editor is not None
self._editor.set_input(line, pos)
[docs]
def set_line_mode(self, line_mode: bool) -> None:
"""Enable/disable input line editing
This method enabled or disables input line editing. When set,
only full lines of input are sent to the session, and each
line of input can be edited before it is sent.
:param line_mode:
Whether or not to process input a line at a time
:type line_mode: `bool`
"""
self._orig_chan.logger.info('%s line editor',
'Enabling' if line_mode else 'Disabling')
assert self._editor is not None
self._editor.set_line_mode(line_mode)
[docs]
def set_echo(self, echo: bool) -> None:
"""Enable/disable echoing of input in line mode
This method enables or disables echoing of input data when
input line editing is enabled.
:param echo:
Whether or not input to echo input as it is entered
:type echo: `bool`
"""
self._orig_chan.logger.info('%s echo',
'Enabling' if echo else 'Disabling')
assert self._editor is not None
self._editor.set_echo(echo)
def write(self, data: str, datatype: DataType = None) -> None:
"""Process data written to the channel"""
if self._editor and datatype is None:
self._editor.process_output(data)
else:
self._orig_chan.write(data, datatype)
class SSHLineEditorSession:
"""Input line editor session wrapper"""
def __init__(self, chan: SSHLineEditorChannel,
orig_session: 'SSHServerSession[str]'):
self._chan = chan
self._orig_session = orig_session
self._editor: Optional[SSHLineEditor] = None
def __getattr__(self, attr: str):
"""Delegate most channel functions to original session"""
return getattr(self._orig_session, attr)
def session_started(self) -> None:
"""Start a session for this newly opened server channel"""
self._editor = self._chan.create_editor()
self._orig_session.session_started()
def terminal_size_changed(self, width: int, height: int,
pixwidth: int, pixheight: int) -> None:
"""The terminal size has changed"""
if self._editor:
self._editor.set_width(width)
self._orig_session.terminal_size_changed(width, height,
pixwidth, pixheight)
def data_received(self, data: str, datatype: DataType) -> None:
"""Process data received from the channel"""
if self._editor:
self._editor.process_input(data, datatype)
else:
self._orig_session.data_received(data, datatype)
def eof_received(self) -> Optional[bool]:
"""Process EOF received from the channel"""
if self._editor:
self._editor.set_line_mode(False)
return self._orig_session.eof_received()