import re import selectors import socket from collections import deque from telnetlib import Telnet from gi.repository import GLib from app.commons import run_task, run_idle, log from app.settings import Settings from app.ui.uicommons import Gtk, Gdk, UI_RESOURCES_PATH, KeyboardKey, MOD_MASK class ExtTelnet(Telnet): def __init__(self, output_callback, **kwargs): super().__init__(**kwargs) self._output_callback = output_callback def interact(self): """Interaction function, emulates a very dumb telnet client.""" with selectors.DefaultSelector() as selector: selector.register(self, selectors.EVENT_READ) while True: for key, events in selector.select(): if key.fileobj is self: try: text = self.read_very_eager() except EOFError as e: msg = "\n*** Connection closed by remote host ***\n" self._output_callback(msg) log(msg) raise e else: if text: self._output_callback(text) class TelnetDialog: """ Dialog of very simple telnet client. """ _COLOR_PATTERN = re.compile("\x1b.*?m") # Color info _ERASING_PATTERN = re.compile("\x1b.*?K") # Erase to right _APP_MODE_PATTERN = re.compile("\x1b.*?(1h)|(1l)") # h - on, l - off _ALL_PATTERN = re.compile(r'(\x1b\[|\x9b)[0-?]*[@-~]') _NOT_SUPPORTED = {"mc", "mcedit", "vi", "nano"} def __init__(self, transient, settings): self._handlers = {"on_profile_changed": self.on_profile_changed, "on_clear": self.on_clear, "on_text_view_realize": self.on_text_view_realize, "on_view_key_press": self.on_view_key_press, "on_info_bar_close": self.on_info_bar_close, "on_connect": self.on_connect, "on_disconnect": self.on_disconnect, "on_close": self.on_close} builder = Gtk.Builder() builder.add_from_file(UI_RESOURCES_PATH + "telnet.glade") builder.connect_signals(self._handlers) self._dialog_window = builder.get_object("dialog_window") self._dialog_window.set_transient_for(transient) self._profile_combo_box = builder.get_object("profile_combo_box") self._info_bar = builder.get_object("info_bar") self._info_message_label = builder.get_object("info_bar_message_label") self._text_view = builder.get_object("text_view") self._buf = builder.get_object("text_buffer") self._end_tag = builder.get_object("end_tag") self._connect_button = builder.get_object("connect_button") self._connect_button.bind_property("visible", builder.get_object("disconnect_button"), "visible", 4) provider = Gtk.CssProvider() provider.load_from_path(UI_RESOURCES_PATH + "style.css") builder.get_object("main_box").get_style_context().add_provider(provider, Gtk.STYLE_PROVIDER_PRIORITY_USER) window_size = settings.get("telnet_dialog_window_size") if window_size: self._dialog_window.resize(*window_size) self._ext_settings = settings self._settings = Settings(settings.settings) self._tn = None self._app_mode = False self._commands = deque(maxlen=10) def show(self): self._dialog_window.show() def on_close(self, window, event): """ Performs shutdown tasks """ self._ext_settings.add("telnet_dialog_window_size", window.get_size()) self.on_disconnect() def on_info_bar_close(self, bar=None, resp=None): self._info_bar.set_visible(False) @run_idle def show_info_message(self, text, message_type): self._info_bar.set_visible(True) self._info_bar.set_message_type(message_type) self._info_message_label.set_text(text) def on_text_view_realize(self, view): self.init_profiles() self.on_connect() @run_idle def init_profiles(self): for p in self._settings.profiles: self._profile_combo_box.append(p, p) self._profile_combo_box.set_active_id(self._settings.current_profile) @run_task def on_connect(self, item=None): try: GLib.idle_add(self._connect_button.set_visible, False) GLib.idle_add(self.on_info_bar_close) user, password = self._settings.user, self._settings.password timeout = self._settings.telnet_timeout self._tn = ExtTelnet(self.append_output, host=self._settings.host, port=self._settings.telnet_port, timeout=timeout) if user != "": self._tn.read_until(b"login: ") self._tn.write(user.encode("utf-8") + b"\n") if password != "": self._tn.read_until(b"Password: ") self._tn.write(password.encode("utf-8") + b"\n") self._tn.interact() except (OSError, EOFError, socket.timeout, ConnectionRefusedError) as e: log("{}: {}".format(self.__class__.__name__, e)) self.show_info_message(str(e), Gtk.MessageType.ERROR) finally: GLib.idle_add(self._connect_button.set_visible, True) @run_task def on_disconnect(self, item=None): if self._tn: GLib.idle_add(self._connect_button.set_visible, True) self._tn.close() def on_profile_changed(self, button): self._settings.current_profile = button.get_active_id() def on_command_done(self, entry): command = entry.get_text() entry.set_text("") if command and self._tn: self._tn.write(command.encode("ascii") + b"\r") def on_clear(self, item=None): self._buf.delete(self._buf.get_start_iter(), self._buf.get_end_iter()) def on_view_key_press(self, view, event): """ Handling keystrokes on press """ if event.keyval == Gdk.KEY_Return: self.do_command() return True key_code = event.hardware_keycode if not KeyboardKey.value_exist(key_code): return key = KeyboardKey(key_code) ctrl = event.state & MOD_MASK if ctrl and key is KeyboardKey.C: if self._tn and self._tn.sock: self._tn.write(b"\x03") # interrupt # last commands navigation if key is KeyboardKey.UP: self.delete_last_command() if self._commands: cmd = self._commands.pop() self._commands.appendleft(cmd) self._buf.insert_at_cursor(cmd, -1) return True elif key is KeyboardKey.DOWN: self.delete_last_command() if self._commands: cmd = self._commands.popleft() self._commands.append(cmd) self._buf.insert_at_cursor(cmd, -1) return True def delete_last_command(self): end = self._buf.get_end_iter() if end.ends_tag(self._end_tag): return if end.backward_to_tag_toggle(self._end_tag): self._buf.delete(self._buf.get_end_iter(), end) def do_command(self): count = self._buf.get_line_count() begin = self._buf.get_iter_at_line(count) end = self._buf.get_end_iter() command = [] while end.backward_to_tag_toggle(self._end_tag): command.append(self._buf.get_text(end, begin, False)) break else: # if buf is empty command.append(self._buf.get_text(begin, end, False)) # to preventing duplication of the command in the buf self._buf.delete(end, begin) if command and self._tn.sock: cmd = command[0] if cmd in self._NOT_SUPPORTED: self.show_info_message("'{}' is not supported by this client.".format(cmd), Gtk.MessageType.ERROR) else: self._tn.write(cmd.encode("ascii") + b"\r") self._commands.append(cmd) @run_idle def append_output(self, txt): t = txt.decode("ascii", errors="ignore") ap = re.search(self._APP_MODE_PATTERN, t) if ap: on, of = ap.group(1), ap.group(2) if on: self._app_mode = True elif of: self._app_mode = False self.on_clear() t = re.sub(self._ALL_PATTERN, "", t) # removing [replacing] ascii escape sequences if self._app_mode: start, end = self._buf.get_start_iter(), self._buf.get_end_iter() count = self._buf.get_line_count() new_lines = t.split("\r\n") ext_lines = self._buf.get_text(start, end, True).split("\r\n") if count < len(new_lines): self._buf.set_text(re.sub(self._ERASING_PATTERN, "", t)) else: for i, line in enumerate(new_lines): if line: ext_lines[i] = re.sub(self._ERASING_PATTERN, "", line) self._buf.set_text("\r\n".join(ext_lines)) else: self._buf.insert_at_cursor(t, -1) insert = self._buf.get_insert() self._text_view.scroll_to_mark(insert, 0.0, True, 0.0, 1.0) self._buf.apply_tag(self._end_tag, self._buf.get_start_iter(), self._buf.get_end_iter()) if __name__ == "__main__": pass