diff --git a/docs/README.md b/docs/README.md index 8a660bb40..c4daea4fb 100644 --- a/docs/README.md +++ b/docs/README.md @@ -10,6 +10,7 @@ platform documentation that versions with the code: | [running-from-source.md](running-from-source.md) | Running from source on Linux: the hybrid (system PyGObject + venv) and pure-venv (pip-built PyGObject) approaches, plus dev/test setup and troubleshooting. | | [INSTALL-macos.md](INSTALL-macos.md) | Installing and running from source on macOS (Homebrew GTK stack). | | [keyboard-shortcuts.md](keyboard-shortcuts.md) | Full shortcut reference for Linux and macOS. | +| [highlighting.md](highlighting.md) | Network-oriented terminal output highlighting profiles and safety behavior. | | [architecture/terminal-input-routing.md](architecture/terminal-input-routing.md) | How pointer and scroll events are shared between VTE and SSH Pilot's own controllers, and where each controller must be attached. Read this before touching terminal controllers. | | [accessibility-automation.md](accessibility-automation.md) | What SSH Pilot exposes over AT-SPI, the rules for accessible names, and what UI automation can and cannot drive through it (Wayland, VTE, drag-and-drop). | | [diagnostics.md](diagnostics.md) | Where the logs live, the crash report, the `--diagnostics` flags and the bug-report bundle. | diff --git a/docs/highlighting.md b/docs/highlighting.md new file mode 100644 index 000000000..408c7668d --- /dev/null +++ b/docs/highlighting.md @@ -0,0 +1,18 @@ +# Terminal output highlighting + +Output highlighting is opt-in under **Settings → Terminal → Highlighting**. +It adds foreground colour without changing visible output bytes. + +The built-in profiles are: + +- **Generic** for common IP addresses, MAC addresses, VLANs, interfaces, and status words. +- **Cisco IOS** for IOS interface and command-output conventions. +- **Junos** for Junos interface and command-output conventions. +- **Huawei VRP** for VRP interface and command-output conventions. + +Highlighting is suspended while a remote application controls foreground or +background colour, reverse or concealed text, or the alternate screen. Parser +state is reset when a terminal reconnects. + +The feature is disabled by default. Select a profile and enable **Highlight +network output** to apply it to new terminal output. diff --git a/po/de.po b/po/de.po index acbe6bb7c..220653ab1 100644 --- a/po/de.po +++ b/po/de.po @@ -17115,6 +17115,48 @@ msgstr "Durchreichen" msgid "Copy key to Server" msgstr "Schlüssel auf Server kopieren" + +#: src/sshpilot/preferences.py:754 src/sshpilot/settings_index.py:123 +#: src/sshpilot/settings_index.py:126 +msgid "Highlighting" +msgstr "" + +#: src/sshpilot/preferences.py:757 src/sshpilot/settings_index.py:123 +msgid "Highlight Network Output" +msgstr "" + +#: src/sshpilot/preferences.py:759 +msgid "Colour IP and MAC addresses, VLANs, interfaces and link status" +msgstr "" + +#: src/sshpilot/preferences.py:768 src/sshpilot/settings_index.py:126 +msgid "Highlight Profile" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Generic" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Cisco IOS" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Junos" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Huawei VRP" +msgstr "" + +#: src/sshpilot/settings_index.py:124 +msgid "syntax;color;colour;cisco;junos;huawei;ip;mac;vlan" +msgstr "" + +#: src/sshpilot/settings_index.py:126 +msgid "cisco;junos;huawei" +msgstr "" + #~ msgid "Connection list actions" #~ msgstr "Aktionen für die Verbindungsliste" diff --git a/po/es.po b/po/es.po index 129a91d85..95bc6685b 100644 --- a/po/es.po +++ b/po/es.po @@ -16991,6 +16991,48 @@ msgstr "paso a través" msgid "Copy key to Server" msgstr "Copiar clave al servidor" + +#: src/sshpilot/preferences.py:754 src/sshpilot/settings_index.py:123 +#: src/sshpilot/settings_index.py:126 +msgid "Highlighting" +msgstr "" + +#: src/sshpilot/preferences.py:757 src/sshpilot/settings_index.py:123 +msgid "Highlight Network Output" +msgstr "" + +#: src/sshpilot/preferences.py:759 +msgid "Colour IP and MAC addresses, VLANs, interfaces and link status" +msgstr "" + +#: src/sshpilot/preferences.py:768 src/sshpilot/settings_index.py:126 +msgid "Highlight Profile" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Generic" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Cisco IOS" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Junos" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Huawei VRP" +msgstr "" + +#: src/sshpilot/settings_index.py:124 +msgid "syntax;color;colour;cisco;junos;huawei;ip;mac;vlan" +msgstr "" + +#: src/sshpilot/settings_index.py:126 +msgid "cisco;junos;huawei" +msgstr "" + #~ msgid "Connection list actions" #~ msgstr "Acciones de la lista de conexiones" diff --git a/po/fa.po b/po/fa.po index 59c4fbdbf..e791da9af 100644 --- a/po/fa.po +++ b/po/fa.po @@ -16746,6 +16746,48 @@ msgstr "عبور مستقیم" msgid "Copy key to Server" msgstr "رونوشت کلید به سرور" + +#: src/sshpilot/preferences.py:754 src/sshpilot/settings_index.py:123 +#: src/sshpilot/settings_index.py:126 +msgid "Highlighting" +msgstr "" + +#: src/sshpilot/preferences.py:757 src/sshpilot/settings_index.py:123 +msgid "Highlight Network Output" +msgstr "" + +#: src/sshpilot/preferences.py:759 +msgid "Colour IP and MAC addresses, VLANs, interfaces and link status" +msgstr "" + +#: src/sshpilot/preferences.py:768 src/sshpilot/settings_index.py:126 +msgid "Highlight Profile" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Generic" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Cisco IOS" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Junos" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Huawei VRP" +msgstr "" + +#: src/sshpilot/settings_index.py:124 +msgid "syntax;color;colour;cisco;junos;huawei;ip;mac;vlan" +msgstr "" + +#: src/sshpilot/settings_index.py:126 +msgid "cisco;junos;huawei" +msgstr "" + #~ msgid "Connection list actions" #~ msgstr "عملیات‌های فهرست اتصال" diff --git a/po/fr.po b/po/fr.po index 7d8cda5dc..fa8c78850 100644 --- a/po/fr.po +++ b/po/fr.po @@ -17101,6 +17101,48 @@ msgstr "Transmission directe" msgid "Copy key to Server" msgstr "Copier la clé sur le serveur" + +#: src/sshpilot/preferences.py:754 src/sshpilot/settings_index.py:123 +#: src/sshpilot/settings_index.py:126 +msgid "Highlighting" +msgstr "" + +#: src/sshpilot/preferences.py:757 src/sshpilot/settings_index.py:123 +msgid "Highlight Network Output" +msgstr "" + +#: src/sshpilot/preferences.py:759 +msgid "Colour IP and MAC addresses, VLANs, interfaces and link status" +msgstr "" + +#: src/sshpilot/preferences.py:768 src/sshpilot/settings_index.py:126 +msgid "Highlight Profile" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Generic" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Cisco IOS" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Junos" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Huawei VRP" +msgstr "" + +#: src/sshpilot/settings_index.py:124 +msgid "syntax;color;colour;cisco;junos;huawei;ip;mac;vlan" +msgstr "" + +#: src/sshpilot/settings_index.py:126 +msgid "cisco;junos;huawei" +msgstr "" + #~ msgid "Connection list actions" #~ msgstr "Actions de la liste des connexions" diff --git a/po/pt.po b/po/pt.po index 4c41726d2..69d5be44e 100644 --- a/po/pt.po +++ b/po/pt.po @@ -16991,6 +16991,48 @@ msgstr "Transmissão direta" msgid "Copy key to Server" msgstr "Copiar chave para o servidor" + +#: src/sshpilot/preferences.py:754 src/sshpilot/settings_index.py:123 +#: src/sshpilot/settings_index.py:126 +msgid "Highlighting" +msgstr "" + +#: src/sshpilot/preferences.py:757 src/sshpilot/settings_index.py:123 +msgid "Highlight Network Output" +msgstr "" + +#: src/sshpilot/preferences.py:759 +msgid "Colour IP and MAC addresses, VLANs, interfaces and link status" +msgstr "" + +#: src/sshpilot/preferences.py:768 src/sshpilot/settings_index.py:126 +msgid "Highlight Profile" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Generic" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Cisco IOS" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Junos" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Huawei VRP" +msgstr "" + +#: src/sshpilot/settings_index.py:124 +msgid "syntax;color;colour;cisco;junos;huawei;ip;mac;vlan" +msgstr "" + +#: src/sshpilot/settings_index.py:126 +msgid "cisco;junos;huawei" +msgstr "" + #~ msgid "Connection list actions" #~ msgstr "Ações da lista de ligações" diff --git a/po/pt_BR.po b/po/pt_BR.po index 887109928..f883f4751 100644 --- a/po/pt_BR.po +++ b/po/pt_BR.po @@ -16916,6 +16916,48 @@ msgstr "Passagem direta" msgid "Copy key to Server" msgstr "Copiar chave para o servidor" + +#: src/sshpilot/preferences.py:754 src/sshpilot/settings_index.py:123 +#: src/sshpilot/settings_index.py:126 +msgid "Highlighting" +msgstr "" + +#: src/sshpilot/preferences.py:757 src/sshpilot/settings_index.py:123 +msgid "Highlight Network Output" +msgstr "" + +#: src/sshpilot/preferences.py:759 +msgid "Colour IP and MAC addresses, VLANs, interfaces and link status" +msgstr "" + +#: src/sshpilot/preferences.py:768 src/sshpilot/settings_index.py:126 +msgid "Highlight Profile" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Generic" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Cisco IOS" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Junos" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Huawei VRP" +msgstr "" + +#: src/sshpilot/settings_index.py:124 +msgid "syntax;color;colour;cisco;junos;huawei;ip;mac;vlan" +msgstr "" + +#: src/sshpilot/settings_index.py:126 +msgid "cisco;junos;huawei" +msgstr "" + #~ msgid "Connection list actions" #~ msgstr "Ações da lista de conexões" diff --git a/po/ru.po b/po/ru.po index 037b96e16..23c72001d 100644 --- a/po/ru.po +++ b/po/ru.po @@ -17027,6 +17027,48 @@ msgstr "Сквозная передача" msgid "Copy key to Server" msgstr "Скопировать ключ на сервер" + +#: src/sshpilot/preferences.py:754 src/sshpilot/settings_index.py:123 +#: src/sshpilot/settings_index.py:126 +msgid "Highlighting" +msgstr "" + +#: src/sshpilot/preferences.py:757 src/sshpilot/settings_index.py:123 +msgid "Highlight Network Output" +msgstr "" + +#: src/sshpilot/preferences.py:759 +msgid "Colour IP and MAC addresses, VLANs, interfaces and link status" +msgstr "" + +#: src/sshpilot/preferences.py:768 src/sshpilot/settings_index.py:126 +msgid "Highlight Profile" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Generic" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Cisco IOS" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Junos" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Huawei VRP" +msgstr "" + +#: src/sshpilot/settings_index.py:124 +msgid "syntax;color;colour;cisco;junos;huawei;ip;mac;vlan" +msgstr "" + +#: src/sshpilot/settings_index.py:126 +msgid "cisco;junos;huawei" +msgstr "" + #~ msgid "Connection list actions" #~ msgstr "Действия со списком подключений" diff --git a/po/zh_CN.po b/po/zh_CN.po index f59a4ccb1..9761da9bb 100644 --- a/po/zh_CN.po +++ b/po/zh_CN.po @@ -16280,6 +16280,48 @@ msgstr "直通" msgid "Copy key to Server" msgstr "复制密钥到服务器" + +#: src/sshpilot/preferences.py:754 src/sshpilot/settings_index.py:123 +#: src/sshpilot/settings_index.py:126 +msgid "Highlighting" +msgstr "" + +#: src/sshpilot/preferences.py:757 src/sshpilot/settings_index.py:123 +msgid "Highlight Network Output" +msgstr "" + +#: src/sshpilot/preferences.py:759 +msgid "Colour IP and MAC addresses, VLANs, interfaces and link status" +msgstr "" + +#: src/sshpilot/preferences.py:768 src/sshpilot/settings_index.py:126 +msgid "Highlight Profile" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Generic" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Cisco IOS" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Junos" +msgstr "" + +#: src/sshpilot/preferences.py:770 +msgid "Huawei VRP" +msgstr "" + +#: src/sshpilot/settings_index.py:124 +msgid "syntax;color;colour;cisco;junos;huawei;ip;mac;vlan" +msgstr "" + +#: src/sshpilot/settings_index.py:126 +msgid "cisco;junos;huawei" +msgstr "" + #~ msgid "Connection list actions" #~ msgstr "连接列表操作" diff --git a/src/sshpilot/core/settings/defaults.py b/src/sshpilot/core/settings/defaults.py index e85c0249d..76efe03fd 100644 --- a/src/sshpilot/core/settings/defaults.py +++ b/src/sshpilot/core/settings/defaults.py @@ -39,6 +39,11 @@ def get_default_config() -> Dict[str, Any]: 'cursor_shape': 'block', # 'system' (follow GTK) | 'on' | 'off'. 'cursor_blink': 'system', + # Colour IPs/MACs/VLANs/interfaces/status in output (see + # terminal_highlight). Profile: 'generic' | 'cisco-ios' | 'junos' + # | 'huawei-vrp'. + 'highlight_enabled': False, + 'highlight_profile': 'generic', 'audible_bell': False, 'term': None, 'pass_through_mode': False, diff --git a/src/sshpilot/preferences.py b/src/sshpilot/preferences.py index 02748bc40..d9f5540c4 100644 --- a/src/sshpilot/preferences.py +++ b/src/sshpilot/preferences.py @@ -35,6 +35,7 @@ normalize_cursor_blink, normalize_cursor_shape, ) +from .terminal_highlight import PROFILES, normalize_profile import gi @@ -748,6 +749,42 @@ def _add_terminal_cursor_group(self, terminal_page): terminal_page.add(cursor_group) + def _add_terminal_highlight_group(self, terminal_page): + """Add the network-output highlighting group.""" + group = Adw.PreferencesGroup(title=_("Highlighting")) + + self.highlight_switch = Adw.SwitchRow() + self.highlight_switch.set_title(_("Highlight Network Output")) + self.highlight_switch.set_subtitle( + _("Colour IP and MAC addresses, VLANs, interfaces and link status") + ) + self.highlight_switch.set_active( + bool(self.config.get_setting('terminal.highlight_enabled', False)) + ) + self.highlight_switch.connect('notify::active', self.on_highlight_enabled_changed) + group.add(self.highlight_switch) + + self.highlight_profile_row = Adw.ComboRow() + self.highlight_profile_row.set_title(_("Highlight Profile")) + profile_model = Gtk.StringList() + for label in (_("Generic"), _("Cisco IOS"), _("Junos"), _("Huawei VRP")): + profile_model.append(label) + self.highlight_profile_row.set_model(profile_model) + current = normalize_profile(self.config.get_setting('terminal.highlight_profile', None)) + self.highlight_profile_row.set_selected(PROFILES.index(current)) + self.highlight_profile_row.connect('notify::selected', self.on_highlight_profile_changed) + group.add(self.highlight_profile_row) + + terminal_page.add(group) + + def on_highlight_enabled_changed(self, switch_row, _param): + self.config.set_setting('terminal.highlight_enabled', switch_row.get_active()) + + def on_highlight_profile_changed(self, combo_row, _param): + index = combo_row.get_selected() + if 0 <= index < len(PROFILES): + self.config.set_setting('terminal.highlight_profile', PROFILES[index]) + def on_cursor_shape_changed(self, combo_row, _param): index = combo_row.get_selected() if not 0 <= index < len(CURSOR_SHAPES): @@ -1257,6 +1294,7 @@ def _build_terminal_preferences_page(self): self._add_terminal_appearance_groups(terminal_page) self._add_terminal_cursor_group(terminal_page) + self._add_terminal_highlight_group(terminal_page) self._add_terminal_backend_group(terminal_page) self._add_terminal_input_groups(terminal_page) self._add_terminal_daemon_group(terminal_page) diff --git a/src/sshpilot/settings_index.py b/src/sshpilot/settings_index.py index 5ec97dc1a..ffab93130 100644 --- a/src/sshpilot/settings_index.py +++ b/src/sshpilot/settings_index.py @@ -119,6 +119,11 @@ def keyword_list(self) -> Tuple[str, ...]: SettingEntry("terminal", N_("Appearance"), N_("Encoding"), N_("charset;utf-8")), SettingEntry("terminal", N_("Cursor"), N_("Cursor Shape"), N_("block;underline;i-beam")), SettingEntry("terminal", N_("Cursor"), N_("Cursor Blinking"), N_("blink")), + SettingEntry( + "terminal", N_("Highlighting"), N_("Highlight Network Output"), + N_("syntax;color;colour;cisco;junos;huawei;ip;mac;vlan"), + ), + SettingEntry("terminal", N_("Highlighting"), N_("Highlight Profile"), N_("cisco;junos;huawei")), SettingEntry("terminal", N_("Backend"), N_("Terminal Backend"), N_("vte;xterm.js")), SettingEntry("terminal", N_("Keyboard"), N_("Terminal Shortcut Pass-through")), SettingEntry("terminal", N_("Keyboard"), N_("Command autocomplete"), diff --git a/src/sshpilot/startup_preload.py b/src/sshpilot/startup_preload.py index f2577757c..414a76168 100644 --- a/src/sshpilot/startup_preload.py +++ b/src/sshpilot/startup_preload.py @@ -84,6 +84,7 @@ "sshpilot.tag_groups", "sshpilot.wol", "sshpilot.terminal_cursor", + "sshpilot.terminal_highlight", "sshpilot.ssh_process_manager", "sshpilot.core.connection_evidence", "sshpilot.terminal_input", diff --git a/src/sshpilot/terminal.py b/src/sshpilot/terminal.py index 36ae7d491..c7cc4a374 100644 --- a/src/sshpilot/terminal.py +++ b/src/sshpilot/terminal.py @@ -48,6 +48,7 @@ # ssh_process_manager.py (GTK-free). Re-exported here so existing # `from .terminal import SSHProcessManager` / `process_manager` callers keep working. from .ssh_process_manager import SSHProcessManager, process_manager # noqa: F401 +from .terminal_highlight import Highlighter, apply_config from .terminal_search import TerminalSearch from .shortcut_utils import ( accel_matches_latin_fallback, @@ -399,6 +400,8 @@ def __init__(self, connection, config, connection_manager, group_color=None): except Exception: self._pass_through_mode = False + self._apply_highlight_settings() + if hasattr(self.config, 'connect'): try: self._config_handler = self.config.connect('setting-changed', self._on_config_setting_changed) @@ -1794,6 +1797,7 @@ def _uninstall_daemon_backend_io(self) -> None: list(tracker.modes), ) tracker.reset() + self._highlighter.reset() return for attr in ('_daemon_commit_handler', '_daemon_size_handler'): handler = getattr(self, attr, None) @@ -1812,6 +1816,7 @@ def _uninstall_daemon_backend_io(self) -> None: list(tracker.modes), ) tracker.reset() + self._highlighter.reset() def _install_daemon_backend_io(self) -> None: """Wire commit/resize through the terminal backend abstraction. @@ -1951,7 +1956,7 @@ def _paint_display(self, data: bytes) -> None: bool(after), list(after), ) - backend.feed(data) + backend.feed(self._highlighter.feed(data)) if tracker is not None and self._needs_legacy_mouse_translation(): # Settles on top of whatever the remote just set, so the emulator # ends up in SGR whenever the remote asked for a legacy mode. @@ -2026,6 +2031,7 @@ def _on_daemon_continuity_lost(self): # text; the damaged remote stream is permanently halted by the # binding and must never resume in this parser state. self.backend.reset(False, True) + self._highlighter.reset() marker = b"\r\n[Earlier terminal output is no longer available]\r\n" self._feed_display(marker) except Exception as e: @@ -5296,8 +5302,30 @@ def _apply_macos_option_key_passthrough(self, enabled: bool) -> bool: backend.set_macos_option_key_passthrough(enabled) return False + @property + def _highlighter(self) -> Highlighter: + # Lazy so widgets built without __init__ (tests, subclasses) still work. + obj = self.__dict__.get("_highlighter_obj") + if obj is None: + obj = self.__dict__["_highlighter_obj"] = Highlighter(enabled=False) + return obj + + def _apply_highlight_settings(self) -> bool: + held = apply_config(self._highlighter, self.config) + logger.debug( + "Output highlighting enabled=%s profile=%s", + self._highlighter.enabled, + self._highlighter.profile, + ) + if held: + # Switched off mid-escape: show the bytes that were held back. + self.backend.feed(held) + return False + def _on_config_setting_changed(self, _config, key, value): - if key == 'terminal.pass_through_mode': + if key in ('terminal.highlight_enabled', 'terminal.highlight_profile'): + GLib.idle_add(self._apply_highlight_settings) + elif key == 'terminal.pass_through_mode': GLib.idle_add(self._apply_pass_through_mode, bool(value)) elif key == 'terminal.encoding': if self._updating_encoding_config: diff --git a/src/sshpilot/terminal_highlight.py b/src/sshpilot/terminal_highlight.py new file mode 100644 index 000000000..5505a6315 --- /dev/null +++ b/src/sshpilot/terminal_highlight.py @@ -0,0 +1,437 @@ +"""Colour network-device terminal output (IPs, MACs, VLANs, interfaces, ...). + +``Highlighter.feed`` rewrites a display byte stream, adding zero-width SGR +escapes around recognised tokens. It uses palette indices (``ESC[31m``.. +``ESC[36m``) so colours follow the terminal theme, and restores the default +foreground with ``ESC[39m`` so no other attribute is disturbed. Visible text +is never changed: stripping SGR from the output gives back the input. + +Streaming rules: + +* Escape sequences split across chunks are held until complete. Control + strings (OSC, DCS, SOS, PM, APC) are passed through by state, not buffered; + OSC ends at BEL or ST, the others at ST only. C1 controls are recognised in + their UTF-8 form (``C2 9B`` ...), never as raw 8-bit bytes, which are UTF-8 + continuation bytes. +* Text is never held back. A token split across chunks is coloured from the + chunk boundary on (the already-shown part stays as it was). +* Text is left alone while the remote has set a foreground/background colour, + reverse or conceal, and while any alternate-screen mode (47/1047/1049) is on. +* A disabled highlighter returns its input unchanged but keeps tracking + terminal state, so enabling it mid-session needs no resync. + +This module is pure Python and does not import ``gi``. +""" + +from __future__ import annotations + +import ipaddress +import re +from typing import Callable, Dict, List, NamedTuple, Optional, Set + +PROFILES = ("generic", "cisco-ios", "junos", "huawei-vrp") + +_RED, _GREEN, _YELLOW, _BLUE, _MAGENTA, _CYAN = 31, 32, 33, 34, 35, 36 + +_MAX_VLAN = 4094 +_MAX_MPLS_LABEL = 1048575 + +Validator = Callable[[bytes, "re.Match[bytes]"], bool] + + +class Rule(NamedTuple): + name: str + pattern: str + colour: int + validate: Optional[Validator] = None + + +def _numbers_at_most(limit: int, low: int = 0) -> Validator: + def check(_full: bytes, m: "re.Match[bytes]") -> bool: + return all(low <= int(n) <= limit for n in re.findall(rb"\d+", m.group())) + + return check + + +def _valid_ipv6(_full: bytes, m: "re.Match[bytes]") -> bool: + address, _, prefix = m.group().partition(b"/") + if prefix and int(prefix) > 128: + return False + if not re.search(rb"[0-9A-Fa-f]", address): + return False + try: + ipaddress.IPv6Address(address.decode("ascii")) + except ValueError: + return False + return True + + +_VERSION_BEFORE = re.compile(rb"(?i)(?:version|ver|release|build)[ :=]+\Z") + + +def _not_a_version(full: bytes, m: "re.Match[bytes]") -> bool: + return _VERSION_BEFORE.search(full, max(0, m.start() - 16), m.start()) is None + + +_OCT = r"(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)" +_BEFORE = r"(?[:-])(?:[0-9A-Fa-f]{2}(?P=ms)){4}" + r"[0-9A-Fa-f]{2}(?![0-9A-Za-z]|:[0-9A-Fa-f])", + _MAGENTA, + ), + Rule( + "mac_dotted", + r"(?[.-])[0-9A-Fa-f]{4}(?P=md)[0-9A-Fa-f]{4}" + r"(?![0-9A-Za-z])", + _MAGENTA, + ), +] +_IP = [ + Rule( + "ipv6", + r"(? (good words, bad words, prefix on bad words, suffix on both) +_EXTRA_STATE = { + "cisco-ios": (r"up|connected|Established", r"down|Idle", "", ""), + "junos": (r"up|Up|Established", r"down|Down|Idle", "", ""), + "huawei-vrp": (r"up|UP", r"down|DOWN", r"[*^]?", r"(?:\([a-z]+\))?"), +} +_ERROR = Rule( + "error", + r"(? List[Rule]: + rules = [*_MAC, *_IP, _IFACE] + if profile in _EXTRA_IFACE: + rules.append(Rule("iface_x", rf"{_BEFORE}{_EXTRA_IFACE[profile]}(?![0-9A-Za-z])", _YELLOW)) + rules += [_VLAN, _MPLS] + if profile == "junos": + rules.append(Rule("mpls_x", rf"{_BEFORE}(?:push|swap)[ \t]+\d{{2,7}}(?![0-9.])", _BLUE, + _numbers_at_most(_MAX_MPLS_LABEL))) + if profile in _EXTRA_STATE: + good, bad, prefix, suffix = _EXTRA_STATE[profile] + rules.append(Rule("state_ok_x", rf"{_WORD}(?:{good}){suffix}{_NOT_WORD}", _GREEN)) + rules.append(Rule("state_bad_x", rf"{_WORD}{prefix}(?:{bad}){suffix}{_NOT_WORD}", _RED)) + rules += _STATE + rules.append(_ERROR) + return rules + + +class _Compiled(NamedTuple): + regex: "re.Pattern[bytes]" + colours: Dict[str, bytes] + validators: Dict[str, Validator] + + +def _compile(profile: str) -> _Compiled: + rules = _rules(profile) + pattern = "|".join(f"(?P<{r.name}>{r.pattern})" for r in rules) + return _Compiled( + re.compile(pattern.encode("ascii")), + {r.name: b"\x1b[%dm" % r.colour for r in rules}, + {r.name: r.validate for r in rules if r.validate}, + ) + + +_COMPILED: Dict[str, _Compiled] = {} + +_NEXT_RE = re.compile(rb"\x1b|\xc2(?:[\x80-\x9f]|\Z)") +_CSI_RE = re.compile(rb"(?:\x1b\[|\xc2\x9b)([0-?]*)([ -/]*)([@-~])") +_STRING_INTRO_RE = re.compile(rb"\x1b([\]PX^_])|\xc2([\x90\x98\x9d\x9e\x9f])") +# Short escapes: anything but the CSI/string introducers [ ] P X ^ _ (ESC) or +# their C1 forms. +_ESC_RE = re.compile(rb"\x1b[ -/]*[0-?@-OQ-WYZ\\`-~]|\xc2[\x80-\x8f\x91-\x97\x99\x9a\x9c]") +_PARTIAL_ESC_RE = re.compile(rb"(?:\x1b(?:\[[0-?]*[ -/]*|[ -/]*)|\xc2(?:\x9b[0-?]*[ -/]*)?)\Z") +# A string ends at ST (and BEL for OSC), is cancelled by CAN/SUB, and is cut +# short by a new ESC (matched without consuming it by the last alternative). +_OSC_END_RE = re.compile(rb"\x07|\x1b\\|\xc2\x9c|[\x18\x1a]|\x1b(?!\\|\Z)") +_ST_END_RE = re.compile(rb"\x1b\\|\xc2\x9c|[\x18\x1a]|\x1b(?!\\|\Z)") +_TAIL_RE = re.compile(rb"[^\x00-\x20\x7f]*\Z") # trailing run of non-break bytes +_ALT_SCREEN_MODES = {b"47", b"1047", b"1049"} +_CARRY_MAX = 64 +_CTX_MAX = 16 # bytes of look-behind kept for _not_a_version across chunks + + +def normalize_profile(value: object) -> str: + return value if isinstance(value, str) and value in PROFILES else "generic" + + +class Highlighter: + """Stateful stream rewriter; one instance per terminal.""" + + def __init__(self, profile: str = "generic", enabled: bool = True) -> None: + self.enabled = enabled + self.set_profile(profile) + self.reset() + + def set_profile(self, profile: str) -> None: + self.profile = normalize_profile(profile) + if self.profile not in _COMPILED: + _COMPILED[self.profile] = _compile(self.profile) + self._compiled = _COMPILED[self.profile] + + def configure(self, enabled: bool, profile: str) -> bytes: + """Apply settings without losing stream state. + + Returns bytes that were held back and must now be displayed unchanged + (non-empty only when switching off mid-escape). + """ + self.set_profile(profile) + flushed = b"" + if self.enabled != enabled: + self._carry = self._ctx = b"" + # The held bytes are on screen after this call either way; keep them + # for parsing (the rest of the sequence may follow) but never re-emit. + flushed = self._hold[self._shown:] if self.enabled else b"" + self._shown = len(self._hold) + self.enabled = enabled + return flushed + + def reset(self) -> None: + """Forget stream state (new session, reattach, lost continuity).""" + self._hold = b"" + self._in_string: Optional[bytes] = None # terminator kind: b"osc" or b"st" + self._alt_modes: Set[bytes] = set() + self._carry = b"" + self._ctx = b"" # plain text before the carry: lookbehind for validators + self._shown = 0 # leading bytes of _hold already displayed + self._reset_sgr() + + def _reset_sgr(self) -> None: + self._fg = self._bg = self._reverse = self._conceal = False + + def feed(self, data: bytes) -> bytes: + if ( + not self.enabled + and not self._hold + and not self._in_string + and b"\x1b" not in data + and b"\xc2" not in data # UTF-8 C1 controls + ): + self._carry = self._ctx = b"" + return data # nothing that changes tracked state: skip the scan + shown, self._shown = self._shown, 0 + buf = self._hold + bytes(data) + self._hold = b"" + out: List[bytes] = [] + i, n = 0, len(buf) + while i < n: + if self._in_string: + end = (_OSC_END_RE if self._in_string == b"osc" else _ST_END_RE).search(buf, i) + if end is None: + if buf.endswith((b"\x1b", b"\xc2")): + self._hold = buf[-1:] + n -= 1 + out.append(buf[i:n]) + break + stop = end.start() if end.group() == b"\x1b" else end.end() + out.append(buf[i:stop]) + i = stop + self._in_string = None + continue + nxt = _NEXT_RE.search(buf, i) + stop = n if nxt is None else nxt.start() + if stop > i: + out.append(self._text(buf[i:stop], last=stop == n)) + i = stop + if nxt is None: + break + self._carry = self._ctx = b"" + i = self._escape(buf, i, out) + if not self.enabled: + self._shown = len(self._hold) # disabled output shows everything + return bytes(data) + result = b"".join(out) + if shown: + if len(self._hold) == len(buf): # still the same unfinished sequence + self._shown = shown + else: + result = result[shown:] # the old hold was displayed already + return result + + def _escape(self, buf: bytes, i: int, out: List[bytes]) -> int: + """Pass one escape sequence through, tracking state; return the new index.""" + m = _CSI_RE.match(buf, i) + if m is not None: + self._csi(m.group(1), m.group(2), m.group(3)) + else: + m = _STRING_INTRO_RE.match(buf, i) + if m is not None: + self._in_string = b"osc" if m.group(1) == b"]" or m.group(2) == b"\x9d" else b"st" + else: + m = _ESC_RE.match(buf, i) + if m is not None: + if m.group() == b"\x1bc": + self.reset() + elif _PARTIAL_ESC_RE.match(buf, i): + self._hold = buf[i:] + return len(buf) + else: + out.append(buf[i:i + 1]) + return i + 1 + out.append(m.group()) + return m.end() + + def _csi(self, params: bytes, inter: bytes, final: bytes) -> None: + if final == b"m" and not inter: + self._sgr(params) + elif final == b"p" and inter == b"!": # DECSTR resets SGR, not the screen + self._reset_sgr() + elif final in (b"h", b"l") and params.startswith(b"?"): + for mode in set(params[1:].split(b";")) & _ALT_SCREEN_MODES: + (self._alt_modes.add if final == b"h" else self._alt_modes.discard)(mode) + + def _sgr(self, params: bytes) -> None: + it = iter(params.split(b";") if params else [b"0"]) + for raw in it: + # Colon sub-parameters (38:2::R:G:B, 4:3) belong to their own + # code: take the code, consume nothing else. + head, colon, _ = raw.partition(b":") + code = int(head) if head.isdigit() else 0 + if code == 0: + self._reset_sgr() + elif code == 7: + self._reverse = True + elif code == 27: + self._reverse = False + elif code == 8: + self._conceal = True + elif code == 28: + self._conceal = False + elif code == 39: + self._fg = False + elif code == 49: + self._bg = False + elif 30 <= code <= 37 or 90 <= code <= 97: + self._fg = True + elif 40 <= code <= 47 or 100 <= code <= 107: + self._bg = True + elif code in (38, 48, 58): + if code == 38: + self._fg = True + elif code == 48: + self._bg = True + if not colon: # semicolon form: 38;5;n / 38;2;r;g;b + kind = next(it, b"") + for _ in range(3 if kind == b"2" else 1 if kind == b"5" else 0): + next(it, None) + + def _text(self, text: bytes, last: bool) -> bytes: + ctx, carry, self._ctx, self._carry = self._ctx, self._carry, b"", b"" + if ( + not self.enabled + or self._alt_modes + or self._fg + or self._bg + or self._reverse + or self._conceal + ): + return text + regex, colours, validators = self._compiled + full = ctx + carry + text + skip = len(ctx) + len(carry) + parts: List[bytes] = [] + pos = 0 + for m in regex.finditer(full): + if m.end() <= skip: + continue + check = validators.get(m.lastgroup) + if check is not None and not check(full, m): + continue + start = max(m.start(), skip) + parts.append(full[max(pos, skip):start]) + parts.append(colours[m.lastgroup]) + parts.append(full[start:m.end()]) + parts.append(b"\x1b[39m") + pos = m.end() + if last: + tail = _TAIL_RE.search(full, max(0, len(full) - _CARRY_MAX - 1)).group() + if len(tail) <= _CARRY_MAX: + self._carry = tail + keep = len(full) - len(tail) + self._ctx = full[max(0, keep - _CTX_MAX):keep] + if not parts: + return text + parts.append(full[max(pos, skip):]) + return b"".join(parts) + + +def apply_config(highlighter: Highlighter, config: object) -> bytes: + """Sync ``highlighter`` with the terminal highlight settings. + + Returns bytes to display unchanged (see ``Highlighter.configure``). + """ + get = config.get_setting # type: ignore[attr-defined] + return highlighter.configure( + get("terminal.highlight_enabled", False) is True, + normalize_profile(get("terminal.highlight_profile", "generic")), + ) diff --git a/tests/test_terminal_highlight.py b/tests/test_terminal_highlight.py new file mode 100644 index 000000000..bc62da089 --- /dev/null +++ b/tests/test_terminal_highlight.py @@ -0,0 +1,805 @@ +"""Network-output highlighter: profiles, chunking, SGR state, pass-through.""" + +import re + +import pytest + +from sshpilot.terminal_highlight import PROFILES, Highlighter, apply_config, normalize_profile + +SGR = re.compile(rb"\x1b\[(?:36|35|34|33|32|31|39)m") + + +def colour(text, code): + return b"\x1b[%dm%s\x1b[39m" % (code, text) + + +def run(profile, *chunks): + h = Highlighter(profile) + return b"".join(h.feed(c) for c in chunks) + + +def strip(data): + return SGR.sub(b"", data) + + +@pytest.mark.parametrize( + "text, code", + [ + (b"192.0.2.1", 36), + (b"192.0.2.0/24", 36), + (b"255.255.255.0", 36), + (b"2001:db8::1", 36), + (b"fe80::1/64", 36), + (b"2001:db8:0:0:0:0:0:1", 36), + (b"02:00:5e:10:00:00", 35), + (b"aa-bb-cc-dd-ee-ff", 35), + (b"0011.2233.4455", 35), + (b"001a-2b3c-4d5e", 35), + (b"vlan 10", 34), + (b"VLAN 10,20,30-40", 34), + (b"VLAN0010", 34), + ], +) +@pytest.mark.parametrize("profile", PROFILES) +def test_generic_tokens_in_every_profile(profile, text, code): + assert run(profile, b"x " + text + b" y") == b"x " + colour(text, code) + b" y" + + +@pytest.mark.parametrize( + "text", + [b"203.0.113.4.5", b"v203.0.113.4", b"256.1.1.1", b"12:34:56", b"std::string", b"abc192.0.2.1", + b"aa:bb:cc:dd:ee", b"192.0.2"], +) +def test_false_positives_left_alone(text): + assert run("generic", text + b"\r\n") == text + b"\r\n" + + +@pytest.mark.parametrize( + "profile, text", + [ + ("cisco-ios", b"GigabitEthernet0/0/1"), + ("cisco-ios", b"Te1/0/48"), + ("cisco-ios", b"Po10"), + ("cisco-ios", b"Vlan100"), + ("cisco-ios", b"Gi0/1.200"), + ("cisco-ios", b"Loopback0"), + ("junos", b"ge-0/0/1"), + ("junos", b"xe-1/2/3:0.100"), + ("junos", b"ae0.0"), + ("junos", b"lo0.0"), + ("junos", b"irb.100"), + ("huawei-vrp", b"GigabitEthernet0/0/1"), + ("huawei-vrp", b"GE0/0/1"), + ("huawei-vrp", b"Eth-Trunk1"), + ("huawei-vrp", b"Vlanif10"), + ("huawei-vrp", b"XGigabitEthernet0/0/1"), + ], +) +def test_interface_names(profile, text): + assert run(profile, b"port " + text + b" ok") == b"port " + colour(text, 33) + b" ok" + + +def test_generic_knows_long_interface_forms_vendors_add_abbreviations(): + # Reworked: the generic profile now carries the shared long forms. + assert run("generic", b"GigabitEthernet0/1") == colour(b"GigabitEthernet0/1", 33) + assert run("generic", b"ge-0/0/1") == colour(b"ge-0/0/1", 33) + assert run("generic", b"Gi0/1") == b"Gi0/1" + assert run("generic", b"so-0/0/0") == b"so-0/0/0" + assert run("cisco-ios", b"Gi0/1") == colour(b"Gi0/1", 33) + assert run("junos", b"so-0/0/0") == colour(b"so-0/0/0", 33) + assert run("junos", b"Gi0/1") == b"Gi0/1" + assert run("huawei-vrp", b"GE0/0/1") == colour(b"GE0/0/1", 33) + assert run("cisco-ios", b"GE0/0/1") == b"GE0/0/1" + + +@pytest.mark.parametrize( + "profile, text, code", + [ + ("cisco-ios", b"up", 32), + ("cisco-ios", b"down", 31), + ("cisco-ios", b"err-disabled", 31), + ("cisco-ios", b"connected", 32), + ("junos", b"Up", 32), + ("junos", b"Down", 31), + ("junos", b"Established", 32), + ("huawei-vrp", b"UP", 32), + ("huawei-vrp", b"up(s)", 32), + ("huawei-vrp", b"*down", 31), + ("huawei-vrp", b"DOWN", 31), + ], +) +def test_link_status(profile, text, code): + assert run(profile, b"is " + text + b",") == b"is " + colour(text, code) + b"," + + +def test_status_words_inside_other_words_are_not_coloured(): + assert run("cisco-ios", b"setup update breakdown up-time") == b"setup update breakdown up-time" + + +def test_generic_status_needs_a_status_context(): + # Reworked: the generic profile colours status, but only in context. + assert run("generic", b"line is up") == b"line is " + colour(b"up", 32) + assert run("generic", b"Link: Down") == b"Link: " + colour(b"Down", 31) + assert run("generic", b"is administratively down") == b"is " + colour( + b"administratively down", 31 + ) + assert run("generic", b"up/up") == colour(b"up/up", 32) + assert run("generic", b"up/down") == colour(b"up/down", 31) + assert run("generic", b"err-disabled") == colour(b"err-disabled", 31) + assert run("generic", b"pick it up") == b"pick it up" + assert run("generic", b"it is up to you") == b"it is up to you" + assert run("generic", b"Gi0/1 up up") == b"Gi0/1 up up" + assert run("cisco-ios", b"x up up") == b"x " + colour(b"up", 32) + b" " + colour(b"up", 32) + + +def test_generic_errors(): + for line in (b"% Invalid input", b"Error: bad", b"syntax error, x", b"unknown command"): + assert run("generic", line) == colour(line, 31) + + +@pytest.mark.parametrize( + "profile, line", + [ + ("cisco-ios", b"% Invalid input detected at '^' marker."), + ("cisco-ios", b"% Incomplete command."), + ("junos", b"syntax error, expecting ."), + ("junos", b"error: device busy"), + ("huawei-vrp", b"Error: Unrecognized command found at '^' position."), + ], +) +def test_error_lines(profile, line): + assert run(profile, line + b"\r\n") == colour(line, 31) + b"\r\n" + + +@pytest.mark.parametrize( + "profile, text", + [("cisco-ios", b"Label 16001"), ("junos", b"push 299792"), ("huawei-vrp", b"In-Label: 1024")], +) +def test_mpls_labels(profile, text): + out = run(profile, b"x " + text + b" y") + assert strip(out) == b"x " + text + b" y" + assert b"\x1b[34m" in out + + +def test_generic_has_mpls_labels_but_push_swap_are_junos_only(): + assert run("generic", b"label 16001") == colour(b"label 16001", 34) + assert run("generic", b"push 299792") == b"push 299792" + assert run("junos", b"push 299792") == colour(b"push 299792", 34) + + +def test_visible_text_unchanged_for_full_session(): + sample = ( + b"R1#show ip int brief\r\nInterface IP-Address Status Protocol\r\n" + b"Gi0/0 198.51.100.1 up up\r\nGi0/1 unassigned administratively down down\r\n" + b"MAC 0011.2233.4455 vlan 10 fe80::1\r\n% Invalid input detected\r\n" + ) + for profile in PROFILES: + assert strip(run(profile, sample)) == sample + + +def test_multiple_tokens_one_chunk(): + out = run("generic", b"192.0.2.1 -> 02:00:5e:10:00:00") + assert out == colour(b"192.0.2.1", 36) + b" -> " + colour(b"02:00:5e:10:00:00", 35) + + +# --- chunking ------------------------------------------------------------- + + +@pytest.mark.parametrize("profile", PROFILES) +def test_every_split_point_preserves_text_and_never_holds_text(profile): + data = ( + b"\x1b[1mhdr\x1b[0m 198.51.100.1/24 02:00:5e:10:00:00 vlan 10 Gi0/1 up\r\n" + b"\x1b]0;title\x07% Invalid input\r\n\x1b[31mred 192.0.2.2\x1b[0m" + ) + for cut in range(len(data) + 1): + h = Highlighter(profile) + first = h.feed(data[:cut]) + second = h.feed(data[cut:]) + assert strip(first + second) == strip(data) + # Only an unfinished escape is held back, never text. + sent = re.sub(rb"\x1b(?:\[[0-?]*[ -/]*)?\Z", b"", data[:cut]) + assert strip(first) == strip(sent) + + +def test_escape_split_across_chunks_is_passed_intact(): + h = Highlighter("generic") + assert h.feed(b"ab\x1b[") == b"ab" + assert h.feed(b"31mred 192.0.2.1") == b"\x1b[31mred 192.0.2.1" + + +def test_split_sgr_still_blocks_highlighting(): + assert run("generic", b"\x1b[3", b"1m192.0.2.1") == b"\x1b[31m192.0.2.1" + + +def test_lone_trailing_esc_is_held_then_released(): + h = Highlighter("generic") + assert h.feed(b"x\x1b") == b"x" + assert h.feed(b"[0m") == b"\x1b[0m" + + +def test_malformed_escape_does_not_stall(): + assert run("generic", b"\x1b[3\nfoo 192.0.2.1")[-12:] == colour(b"192.0.2.1", 36)[-12:] + + +def test_token_split_across_chunks_colours_the_continuation(): + out = run("generic", b"ip 192.0.2.", b"1 up") + assert out == b"ip 192.0.2." + colour(b"1", 36) + b" up" + + +def test_token_split_before_any_match_is_possible(): + out = run("generic", b"02:00:5e:", b"10:00:00") + assert strip(out) == b"02:00:5e:10:00:00" + assert out.endswith(colour(b"10:00:00", 35)) + + +def test_prompt_without_newline_is_emitted_immediately(): + h = Highlighter("cisco-ios") + assert h.feed(b"R1(config-if)#") == b"R1(config-if)#" + assert h.feed(b"ip address 192.0.2.1") == b"ip address " + colour(b"192.0.2.1", 36) + + +def test_carry_resets_at_whitespace(): + out = run("generic", b"foo", b" 192.0.2.1") + assert out == b"foo " + colour(b"192.0.2.1", 36) + + +def test_osc_string_split_across_chunks_is_untouched(): + h = Highlighter("generic") + out = h.feed(b"\x1b]52;c;") + h.feed(b"192.0.2.1 02:00:5e:10:00:00") + h.feed(b"\x07 192.0.2.2") + assert out == b"\x1b]52;c;192.0.2.1 02:00:5e:10:00:00\x07 " + colour(b"192.0.2.2", 36) + + +def test_osc_terminated_by_st_split_between_esc_and_backslash(): + h = Highlighter("generic") + out = h.feed(b"\x1b]0;192.0.2.1\x1b") + h.feed(b"\\ 192.0.2.2") + assert out == b"\x1b]0;192.0.2.1\x1b\\ " + colour(b"192.0.2.2", 36) + + +# --- remote SGR ----------------------------------------------------------- + + +@pytest.mark.parametrize( + "sgr", [b"\x1b[31m", b"\x1b[92m", b"\x1b[41m", b"\x1b[38;5;196m", b"\x1b[38;2;1;2;3m", + b"\x1b[48;5;1m", b"\x1b[38:2::1:2:3m", b"\x1b[7m", b"\x1b[8m", b"\x1b[1;32m"] +) +def test_remote_colour_is_not_clobbered(sgr): + data = sgr + b"192.0.2.1 up" + assert run("cisco-ios", data) == data + + +@pytest.mark.parametrize( + "reset", [b"\x1b[0m", b"\x1b[m", b"\x1b[39m", b"\x1b[49m", b"\x1b[27m", b"\x1b[28m"] +) +def test_highlighting_resumes_after_remote_resets(reset): + # Each reset only matters for the attribute it undoes. + setter = {b"\x1b[39m": b"\x1b[31m", b"\x1b[49m": b"\x1b[41m", b"\x1b[27m": b"\x1b[7m", + b"\x1b[28m": b"\x1b[8m"}.get(reset, b"\x1b[31m") + out = run("generic", setter + b"x" + reset + b"192.0.2.1") + assert out.endswith(colour(b"192.0.2.1", 36)) + + +def test_bold_and_underline_do_not_block_highlighting(): + out = run("generic", b"\x1b[1;4m192.0.2.1") + assert out == b"\x1b[1;4m" + colour(b"192.0.2.1", 36) + + +def test_remote_colour_survives_across_chunks_and_lines(): + h = Highlighter("generic") + h.feed(b"\x1b[33m") + assert h.feed(b"192.0.2.1\r\n192.0.2.2") == b"192.0.2.1\r\n192.0.2.2" + + +def test_ris_and_decstr_reset_state(): + assert run("generic", b"\x1b[31m\x1bc192.0.2.1") == b"\x1b[31m\x1bc" + colour(b"192.0.2.1", 36) + assert run("generic", b"\x1b[31m\x1b[!p192.0.2.1").endswith(colour(b"192.0.2.1", 36)) + + +def test_injected_colour_restores_default_foreground_only(): + out = run("generic", b"\x1b[1m192.0.2.1") + assert b"\x1b[39m" in out and b"\x1b[0m" not in out + + +# --- alt screen ----------------------------------------------------------- + + +@pytest.mark.parametrize("mode", [b"1049", b"47", b"1047", b"25;1049"]) +def test_alt_screen_is_skipped_then_resumes(mode): + h = Highlighter("generic") + assert h.feed(b"\x1b[?" + mode + b"h192.0.2.1") == b"\x1b[?" + mode + b"h192.0.2.1" + assert h.feed(b"\x1b[?" + mode + b"l192.0.2.1") == b"\x1b[?" + mode + b"l" + colour( + b"192.0.2.1", 36 + ) + + +def test_non_alt_private_modes_do_not_disable(): + assert run("generic", b"\x1b[?25l192.0.2.1").endswith(colour(b"192.0.2.1", 36)) + + +def test_alt_screen_mode_split_across_chunks(): + h = Highlighter("generic") + h.feed(b"\x1b[?10") + h.feed(b"49h") + assert h.feed(b"192.0.2.1") == b"192.0.2.1" + + +# --- passthrough / settings ---------------------------------------------- + + +def test_bytes_without_esc_or_tokens_pass_through_exactly(): + data = bytes(b for b in range(256) if b != 0x1B) * 4 + assert run("cisco-ios", data) == data + + +def test_utf8_split_mid_character(): + text = "café 192.0.2.1".encode() + out = run("generic", text[:4], text[4:]) + assert strip(out) == text + + +def test_empty_feed(): + assert Highlighter().feed(b"") == b"" + + +def test_profile_fallback(): + assert normalize_profile("junos") == "junos" + assert normalize_profile("bogus") == "generic" + assert normalize_profile(None) == "generic" + + +def test_full_size_chunk_keeps_visible_text(): + line = b"Gi0/1 is up, 192.0.2.1 02:00:5e:10:00:00 vlan 10\r\n" + data = line * (32768 // len(line)) + out = run("cisco-ios", data) + assert strip(out) == data + + +# --- TerminalWidget hook --------------------------------------------------- + + +def _widget(highlighter): + import types + + from sshpilot.terminal import TerminalWidget + + fed = [] + widget = types.SimpleNamespace( + backend=types.SimpleNamespace(feed=fed.append), + _highlighter=highlighter, + _mouse_tracking=None, + ) + return TerminalWidget._paint_display.__get__(widget), fed + + +def test_paint_display_feeds_highlighted_bytes(): + paint, fed = _widget(Highlighter("generic")) + paint(b"host 192.0.2.1") + assert fed == [b"host " + colour(b"192.0.2.1", 36)] + + +def test_paint_display_disabled_is_byte_identical(): + paint, fed = _widget(Highlighter("generic", enabled=False)) + paint(b"host 192.0.2.1") + assert fed == [b"host 192.0.2.1"] + + +# --- regression: control strings, C1, state, validation --------------------- + +PLAIN_IP = colour(b"192.0.2.2", 36) + + +@pytest.mark.parametrize("intro", [b"\x1bP", b"\x1bX", b"\x1b^", b"\x1b_"]) +def test_bel_does_not_end_dcs_sos_pm_apc(intro): + data = intro + b"q 192.0.2.1\x07 192.0.2.1 02:00:5e:10:00:00" + h = Highlighter("generic") + assert h.feed(data) == data + assert h.feed(b"\x1b\\ 192.0.2.2") == b"\x1b\\ " + PLAIN_IP + + +def test_bel_ends_osc_only(): + assert run("generic", b"\x1b]0;192.0.2.1\x07 192.0.2.2") == b"\x1b]0;192.0.2.1\x07 " + PLAIN_IP + + +@pytest.mark.parametrize( + "chunks", + [ + [b"\xc2\x9d0;192.0.2.1\xc2\x9c 192.0.2.2"], + [b"\xc2\x9d0;192.0.2.1\xc2", b"\x9c 192.0.2.2"], + [b"\xc2", b"\x9d0;192.0.2.1\xc2\x9c 192.0.2.2"], + [b"\xc2\x90q192.0.2.1\x07\xc2\x9c 192.0.2.2"], + ], +) +def test_c1_strings_are_passed_through_untouched(chunks): + out = run("generic", *chunks) + assert out.startswith(b"".join(chunks)[:-len(b" 192.0.2.2")]) + assert out.endswith(b" " + PLAIN_IP) + assert out.count(b"\x1b[36m") == 1 + + +def test_c1_csi_sgr_is_tracked_and_not_clobbered(): + assert run("generic", b"\xc2\x9b31m192.0.2.1") == b"\xc2\x9b31m192.0.2.1" + assert run("generic", b"\xc2\x9b31m", b"192.0.2.1") == b"\xc2\x9b31m192.0.2.1" + assert run("generic", b"\xc2", b"\x9b31m192.0.2.1") == b"\xc2\x9b31m192.0.2.1" + + +def test_c1_alt_screen_and_raw_8bit_bytes(): + assert run("generic", b"\xc2\x9b?1049h192.0.2.1") == b"\xc2\x9b?1049h192.0.2.1" + # A raw 0x9b is a UTF-8 continuation byte (here "ě"), not CSI. + assert run("generic", "ě 192.0.2.1".encode()) == "ě ".encode() + colour(b"192.0.2.1", 36) + + +def test_disable_flushes_held_partial_escape_exactly(): + h = Highlighter("generic") + assert h.feed(b"a\x1b[3") == b"a" + assert h.configure(False, "generic") == b"\x1b[3" + # The rest of that sequence now arrives while disabled: byte-identical. + assert h.feed(b"1m192.0.2.1") == b"1m192.0.2.1" + assert h.configure(True, "generic") == b"" + # State survived the round trip: the remote's red is still in force. + assert h.feed(b"192.0.2.1") == b"192.0.2.1" + + +def test_disabled_feed_is_byte_identical_for_every_split(): + data = b"\x1b[1mhdr\x1b[0m 192.0.2.1 \x1b]0;t\x07 02:00:5e:10:00:00\r\n\x1b[?1049h" + for cut in range(len(data) + 1): + h = Highlighter("cisco-ios", enabled=False) + assert h.feed(data[:cut]) + h.feed(data[cut:]) == data + + +def test_state_survives_toggle_and_profile_change(): + h = Highlighter("generic") + h.feed(b"\x1b[?1049h") + assert h.configure(False, "generic") == b"" + assert h.configure(True, "cisco-ios") == b"" + assert h.profile == "cisco-ios" + assert h.feed(b"192.0.2.1") == b"192.0.2.1" # still on the alt screen + h.feed(b"\x1b[?1049l") + assert h.feed(b"192.0.2.1") == colour(b"192.0.2.1", 36) + + +def test_state_is_tracked_while_disabled(): + h = Highlighter("generic", enabled=False) + h.feed(b"\x1b[?1049h\x1b]0;unterminated") + h.configure(True, "generic") + assert h.feed(b"192.0.2.1 ") == b"192.0.2.1 " # inside the OSC string + assert h.feed(b"\x07\x1b[?1049l192.0.2.1") == b"\x07\x1b[?1049l" + colour(b"192.0.2.1", 36) + + +def test_reset_clears_hold_string_and_alt_screen(): + h = Highlighter("generic") + h.feed(b"\x1b[?1049h\x1b]0;x\x1b") + h.reset() + assert h.feed(b"192.0.2.1") == colour(b"192.0.2.1", 36) + h.feed(b"x\x1b[3") + h.reset() + assert h.feed(b"1m 192.0.2.1") == b"1m " + colour(b"192.0.2.1", 36) + + +def test_reset_keeps_settings(): + h = Highlighter("junos") + h.reset() + assert (h.enabled, h.profile) == (True, "junos") + + +class _Config: + def __init__(self, **values): + self.values = values + + def get_setting(self, key, default=None): + return self.values.get(key, default) + + +def test_apply_config(): + h = Highlighter(enabled=False) + apply_config(h, _Config(**{"terminal.highlight_enabled": True, + "terminal.highlight_profile": "huawei-vrp"})) + assert (h.enabled, h.profile) == (True, "huawei-vrp") + apply_config(h, _Config(**{"terminal.highlight_enabled": "yes", + "terminal.highlight_profile": "nope"})) + assert (h.enabled, h.profile) == (False, "generic") + + +@pytest.mark.parametrize( + "text", + [b"::ffff:203.0.113.4", b"1::", b"2001:db8::1/128", b"fe80::1/64", b"dead:beef::1"], +) +def test_ipv6_valid_forms(text): + assert run("generic", b"a " + text + b" b") == b"a " + colour(text, 36) + b" b" + + +@pytest.mark.parametrize( + "text", + [b"1:2:3:4:5:6:7:8:9", b"2001:db8::1/129", b"1:2:3", b"12:34:56", b"::g", b":::", + b"1::2::3", b"2001:db8:0:0:0:0:0:0:1", b"::ffff:1.2.3.999"], +) +def test_ipv6_invalid_or_overlong_forms_are_plain(text): + assert run("generic", b"a " + text + b" b") == b"a " + text + b" b" + + +@pytest.mark.parametrize( + "line", + [b"version 203.0.113.4", b"Version: 203.0.113.4", b"ver 192.0.2.1", b"Release 198.51.100.9"], +) +def test_version_like_ipv4_is_plain(line): + assert run("cisco-ios", line) == line + + +def test_ipv4_prefix_length_is_range_checked(): + assert run("generic", b"192.0.2.1/32") == colour(b"192.0.2.1/32", 36) + assert run("generic", b"192.0.2.1/33") == colour(b"192.0.2.1", 36) + b"/33" + + +@pytest.mark.parametrize( + "text, ok", + [(b"vlan 1", True), (b"vlan 4094", True), (b"vlan 4095", False), (b"vlan 0", False), + (b"vlan 10,20,4094", True), (b"vlan 10,5000", False), (b"VLAN4094", True), + (b"VLAN4095", False), (b"vlan-id 99", True)], +) +def test_vlan_range(text, ok): + assert run("generic", text) == (colour(text, 34) if ok else text) + + +@pytest.mark.parametrize( + "text, ok", + [(b"label 1048575", True), (b"label 1048576", False), (b"Label: 16", True), + (b"label 9999999", False)], +) +def test_mpls_label_range(text, ok): + assert run("generic", text) == (colour(text, 34) if ok else text) + + +@pytest.mark.parametrize( + "text", + [b"aa:bb-cc:dd-ee:ff", b"aa-bb:cc-dd:ee-ff", b"aabb.ccdd-eeff", b"aabb-ccdd.eeff"], +) +def test_mixed_mac_separators_are_plain(text): + assert run("generic", text) == text + + +def test_mac_is_not_taken_from_inside_an_eight_group_ipv6(): + text = b"02:00:5e:10:00:00:00:11" + assert run("generic", text) == colour(text, 36) + + +@pytest.mark.parametrize( + "setup, still_off", + [ + ([b"\x1b[?1049h", b"\x1b[?47h"], b"\x1b[?1049l"), + ([b"\x1b[?47h", b"\x1b[?1047h"], b"\x1b[?47l"), + ([b"\x1b[?1047;1049h"], b"\x1b[?1047l"), + ], +) +def test_alt_screen_modes_are_a_set(setup, still_off): + h = Highlighter("generic") + for chunk in setup: + h.feed(chunk) + h.feed(still_off) + assert h.feed(b"192.0.2.1") == b"192.0.2.1" + for off in (b"\x1b[?47l", b"\x1b[?1047l", b"\x1b[?1049l"): + h.feed(off) + assert h.feed(b"192.0.2.1") == colour(b"192.0.2.1", 36) + + +# --- widget lifecycle --------------------------------------------------------- + + +def _lifecycle_widget(config=None): + import types + + fed = [] + return types.SimpleNamespace( + backend=types.SimpleNamespace(feed=fed.append, reset=lambda *a: None), + _highlighter=Highlighter("generic"), + _mouse_tracking=None, + config=config, + ), fed + + +def test_widget_continuity_loss_resets_highlighter_before_marker(): + from sshpilot.terminal import TerminalWidget + + widget, fed = _lifecycle_widget() + widget._feed_display = lambda data: TerminalWidget._paint_display(widget, data) + widget._highlighter.feed(b"\x1b[?1049h\x1b[3") + TerminalWidget._on_daemon_continuity_lost(widget) + marker = b"\r\n[Earlier terminal output is no longer available]\r\n" + assert fed == [marker] + assert widget._highlighter.feed(b"192.0.2.1") == colour(b"192.0.2.1", 36) + + +def test_widget_uninstall_daemon_io_resets_highlighter(): + from sshpilot.terminal import TerminalWidget + + widget, _ = _lifecycle_widget() + widget._uninstall_selection_feed_pause = lambda: None + widget._highlighter.feed(b"\x1b[?1049h") + TerminalWidget._uninstall_daemon_backend_io(widget) + assert widget._highlighter.feed(b"192.0.2.1") == colour(b"192.0.2.1", 36) + + +def test_widget_disable_flushes_held_bytes_to_backend(): + from sshpilot.terminal import TerminalWidget + + config = _Config(**{"terminal.highlight_enabled": True}) + widget, fed = _lifecycle_widget(config) + widget._highlighter.feed(b"x\x1b[3") + config.values["terminal.highlight_enabled"] = False + TerminalWidget._apply_highlight_settings(widget) + assert fed == [b"\x1b[3"] + + +# --- regression: colon sub-parameters, string aborts, partials ----------------- + + +def test_colon_truecolour_sets_fg_and_does_not_leak_subparameters(): + # 38:2::R:G:B has an empty colorspace slot; none of R/G/B=0 may reset fg. + data = b"\x1b[38:2::255:128:0m192.0.2.1" + assert run("generic", data) == data + data = b"\x1b[38:2::0:0:0m192.0.2.1" + assert run("generic", data) == data + + +def test_colon_background_and_underline_colour(): + assert run("generic", b"\x1b[48:2::1:2:3m192.0.2.1") == b"\x1b[48:2::1:2:3m192.0.2.1" + # 58 (underline colour) never blocks highlighting, in either form. + for sgr in (b"\x1b[58:2::1:2:3m", b"\x1b[58;2;1;2;3m", b"\x1b[58;5;1m"): + assert run("generic", sgr + b"192.0.2.1") == sgr + colour(b"192.0.2.1", 36) + + +def test_colon_subparameter_of_another_code_does_not_reset(): + data = b"\x1b[31;4:0m192.0.2.1" + assert run("generic", data) == data + assert run("generic", b"\x1b[4:3m192.0.2.1") == b"\x1b[4:3m" + colour(b"192.0.2.1", 36) + + +def test_semicolon_extended_colours_still_consume_their_arguments(): + # 38;5;0 must not let the trailing 0 reset; a following 1 is bold, not colour. + assert run("generic", b"\x1b[38;5;0m192.0.2.1") == b"\x1b[38;5;0m192.0.2.1" + assert run("generic", b"\x1b[38;2;0;0;0m192.0.2.1") == b"\x1b[38;2;0;0;0m192.0.2.1" + out = run("generic", b"\x1b[48;5;0;1m192.0.2.1") + assert out == b"\x1b[48;5;0;1m192.0.2.1" + + +@pytest.mark.parametrize("cancel", [b"\x18", b"\x1a"]) +@pytest.mark.parametrize("intro", [b"\x1b]0;", b"\x1bP", b"\x1b_"]) +def test_can_and_sub_end_control_strings(intro, cancel): + assert run("generic", intro + b"x" + cancel + b"192.0.2.1") == ( + intro + b"x" + cancel + colour(b"192.0.2.1", 36) + ) + + +def test_new_esc_ends_control_string_without_being_consumed(): + out = run("generic", b"\x1b]0;x\x1b[31m192.0.2.1") + assert out == b"\x1b]0;x\x1b[31m192.0.2.1" # the CSI was parsed: red is in force + out = run("generic", b"\x1b]0;x\x1b[0m192.0.2.1") + assert out == b"\x1b]0;x\x1b[0m" + colour(b"192.0.2.1", 36) + + +def test_esc_split_inside_string_terminator_is_still_st(): + assert run("generic", b"\x1bPq\x1b", b"\\192.0.2.1") == b"\x1bPq\x1b\\" + colour( + b"192.0.2.1", 36 + ) + + +def test_esc_with_intermediate_is_held_until_final(): + h = Highlighter("generic") + assert h.feed(b"a\x1b(") == b"a" + assert h.feed(b"B192.0.2.1") == b"\x1b(B" + colour(b"192.0.2.1", 36) + + +@pytest.mark.parametrize("text, ok", [ + (b"implicit-null", True), (b"explicit-null", True), + (b"nonimplicit-null", False), (b"implicit-nullable", False), (b"x-implicit-null", False), +]) +def test_null_labels_need_word_boundaries(text, ok): + assert run("generic", text) == (colour(text, 34) if ok else text) + + +def test_overlong_trailing_token_is_not_carried(): + h = Highlighter("generic") + h.feed(b"x" * 100 + b"192.0.2.") + assert h.feed(b"1") == b"1" # carry dropped: no half-token continuation + h.feed(b"ip 192.0.2.") + assert h.feed(b"1") == colour(b"1", 36) + + +def test_disabled_esc_free_chunk_returns_same_object_without_regex(monkeypatch): + import sshpilot.terminal_highlight as th + + class Boom: + def __getattr__(self, name): + raise AssertionError("regex used for an ESC-free chunk while disabled") + + h = Highlighter(enabled=False) + for name in ("_NEXT_RE", "_CSI_RE", "_ESC_RE", "_PARTIAL_ESC_RE", "_TAIL_RE"): + monkeypatch.setattr(th, name, Boom()) + data = b"Gi0/1 192.0.2.1 is up\r\n" + assert h.feed(data) is data + + +def test_disabled_still_tracks_alt_screen_across_esc_chunks(): + h = Highlighter(enabled=False) + assert h.feed(b"\x1b[?1049h") == b"\x1b[?1049h" + assert h.feed(b"plain") == b"plain" + h.configure(True, "generic") + assert h.feed(b"192.0.2.1") == b"192.0.2.1" # still on the alternate screen + + +def test_widget_highlighter_is_constructed_once_per_widget(monkeypatch): + import sshpilot.terminal as terminal_mod + + built = [] + real = terminal_mod.Highlighter + monkeypatch.setattr( + terminal_mod, "Highlighter", lambda *a, **k: built.append(1) or real(*a, **k) + ) + w = terminal_mod.TerminalWidget.__new__(terminal_mod.TerminalWidget) + for _ in range(5): + w._highlighter + assert len(built) == 1 + + +def test_highlight_module_is_preloaded(): + from sshpilot import startup_preload + + assert "sshpilot.terminal_highlight" in startup_preload.MODULES + + +def test_enabling_mid_escape_keeps_the_pending_sequence(): + h = Highlighter(enabled=False) + assert h.feed(b"\x1b") == b"\x1b" + assert h.configure(True, "generic") == b"" + # The ESC is already on screen: only the rest is emitted, red stays in force. + assert h.feed(b"[31m 192.0.2.1") == b"[31m 192.0.2.1" + + +def test_enabling_mid_string_keeps_the_pending_string(): + h = Highlighter(enabled=False) + assert h.feed(b"\x1b]0;ti") == b"\x1b]0;ti" + h.configure(True, "generic") + assert h.feed(b"192.0.2.1\x07 192.0.2.2") == b"192.0.2.1\x07 " + colour(b"192.0.2.2", 36) + + +def test_enabling_mid_escape_still_unfinished_never_replays(): + h = Highlighter(enabled=False) + h.feed(b"\x1b") + h.configure(True, "generic") + assert h.feed(b"[") == b"" + assert h.feed(b"31m 192.0.2.1") == b"[31m 192.0.2.1" + + +def test_disabling_after_that_flushes_only_unshown_bytes(): + h = Highlighter(enabled=False) + h.feed(b"\x1b") + h.configure(True, "generic") + h.feed(b"[") + assert h.configure(False, "generic") == b"[" + + +def test_disabled_chunk_clears_the_carry(): + h = Highlighter() + assert h.feed(b"192.0.2.") == b"192.0.2." + h.configure(False, "generic") + assert h.feed(b"break ") == b"break " + h.configure(True, "generic") + assert h.feed(b"1") == b"1" + + +def test_version_guard_works_across_chunks(): + h = Highlighter() + assert h.feed(b"version ") == b"version " + assert h.feed(b"203.0.113.4") == b"203.0.113.4" + h.reset() + assert h.feed(b"host ") == b"host " + assert h.feed(b"203.0.113.4") == colour(b"203.0.113.4", 36) + + +def test_apply_highlight_settings_logs_state_at_debug(caplog): + import logging + + from sshpilot.terminal import TerminalWidget + + w = TerminalWidget.__new__(TerminalWidget) + w.config = _Config(**{"terminal.highlight_enabled": True, "terminal.highlight_profile": "junos"}) + with caplog.at_level(logging.DEBUG, logger="sshpilot.terminal"): + TerminalWidget._apply_highlight_settings(w) + assert "Output highlighting enabled=True profile=junos" in caplog.text