From c00deef3d77e55ef90084396e0c91e1fe9b13da0 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 23 Oct 2026 18:58:49 +0200 Subject: [PATCH] fix: keep systemd's OSC 3008 shell context out of command output (#41) Backport of the 0.15.1 fix onto 0ba5081, the commit netOrk pins, which cannot take 0.15.x before it moves to napalm-device-types 4.22. systemd 258 (Ubuntu 26.04) ships /etc/profile.d/80-systemd-osc-context.sh: an interactive bash wraps every command in OSC 3008 sequences from PS0 and PROMPT_COMMAND, each with a fresh UUID. They ended up in every value read through netmiko's shell, and because the prompt never looked the same twice, the prompt line stayed in the output too. os_version grew past 255 characters and netOrk could not store the poll. The session now unsets PS0 and PROMPT_COMMAND right after switching the history off and reads its prompt anew; _send() strips terminal codes from what it returns. --- napalm_linux/linux.py | 22 ++++++- tests/test_sudo_off_the_command_line.py | 3 +- tests/test_systemd_osc_context.py | 78 +++++++++++++++++++++++++ 3 files changed, 99 insertions(+), 4 deletions(-) create mode 100644 tests/test_systemd_osc_context.py diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 7d5f827..6cc5036 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -167,6 +167,12 @@ _PRIVILEGED_STDIN = ( #: a shell without ``set -o history`` (dash) only complains, to /dev/null. _HISTORY_OFF = " unset HISTFILE; set +o history 2>/dev/null; true" +#: Sent right after ``_HISTORY_OFF`` (#41): systemd 258's +#: ``/etc/profile.d/80-systemd-osc-context.sh`` wraps every command in OSC 3008 +#: sequences from ``PS0`` and ``PROMPT_COMMAND``, each with a fresh UUID. It appends +#: to ``PROMPT_COMMAND`` as an array, so only ``unset`` removes its hook. +_PROMPT_HOOKS_OFF = " unset PS0 PROMPT_COMMAND; true" + #: First in every script ``_sudo`` runs: a sudo that does not prompt (NOPASSWD, #: a cached timestamp) leaves the password on stdin, and nothing may read it. _NO_STDIN = "exec /dev/null || true", expect_string=r"[#$>]\s*$") @@ -430,15 +445,16 @@ class LinuxDriver( # ------------------------------------------------------------------ def _send(self, command: str, read_timeout: float = 100) -> str: - """Send a command and return stripped output.""" + """Send a command and return its output without terminal codes, stripped.""" if not self._device: raise ConnectionClosedException("Not connected") - return self._device.send_command( + output = self._device.send_command( command, read_timeout=read_timeout, cmd_verify=False, expect_string=_expect_for(command), - ).strip() + ) + return strip_terminal_codes(output).strip() def _sudo(self, command: str, read_timeout: float = 100) -> str: """Run *command* as root and return its output, stdout and stderr together. diff --git a/tests/test_sudo_off_the_command_line.py b/tests/test_sudo_off_the_command_line.py index 98b07d3..5e15be6 100644 --- a/tests/test_sudo_off_the_command_line.py +++ b/tests/test_sudo_off_the_command_line.py @@ -236,7 +236,8 @@ class TestHistoryOff: def test_a_shell_that_refuses_it_does_not_stop_the_connection(self): driver = LinuxDriver("h", "admin", "p") device = MagicMock() - device.send_command.side_effect = [Exception("no prompt"), "", "/usr/bin/apt"] + # history off (refused), prompt hooks off (#41), stty, package manager + device.send_command.side_effect = [Exception("no prompt"), "", "", "/usr/bin/apt"] with patch("napalm_linux.linux.ConnectHandler", return_value=device): driver.open() diff --git a/tests/test_systemd_osc_context.py b/tests/test_systemd_osc_context.py new file mode 100644 index 0000000..0ed9bb3 --- /dev/null +++ b/tests/test_systemd_osc_context.py @@ -0,0 +1,78 @@ +"""systemd's shell context sequences stay out of every command's output (#41). + +systemd 258 (Ubuntu 26.04) ships ``/etc/profile.d/80-systemd-osc-context.sh``. +An interactive bash then wraps every command in OSC 3008 sequences, from ``PS0`` +before it runs and from ``PROMPT_COMMAND`` before the next prompt, each with a +fresh UUID. netmiko's interactive shell is such a bash: the sequences ended up in +every parsed value, and since the prompt never looked the same twice, netmiko +left the prompt line in the output as well. ``os_version`` grew past 255 +characters and netOrk could not store the poll. +""" + +from unittest.mock import MagicMock, patch + +from napalm_linux.linux import _HISTORY_OFF, _PROMPT_HOOKS_OFF, LinuxDriver + +#: What ``cat /etc/os-release | grep PRETTY_NAME ...`` returned through netmiko on +#: an Ubuntu 26.04.1 host (2026-10-23), the prompt line already gone. +_UBUNTU_2604_OS_RELEASE = ( + "\x1b]3008;end=87c2e1c5-f850-4429-9f06-749e9be783e6;exit=success\x1b\\" + "\x1b]3008;start=bc3bc23b-2652-4826-8192-be447f97dd6f;type=command;" + "machineid=35cf42f3beb446bbb9aa95b937a0cf48;user=netork;hostname=ubuntu;" + "bootid=d32bd88c-dfce-40fe-9869-257bf50ef0e0;pid=20059;cwd=/home/netork\x1b\\" + "Ubuntu 26.04.1 LTS\n" +) + + +def _opened(device: MagicMock) -> LinuxDriver: + driver = LinuxDriver("h", "admin", "p") + with patch("napalm_linux.linux.ConnectHandler", return_value=device): + driver.open() + return driver + + +class TestPromptHooksOff: + def test_the_session_drops_the_hooks_right_after_the_history(self): + device = MagicMock() + device.send_command.return_value = "/usr/bin/apt" + + _opened(device) + + sent = [c.args[0] for c in device.send_command.call_args_list] + assert sent[:2] == [_HISTORY_OFF, _PROMPT_HOOKS_OFF] + + def test_it_unsets_both_hooks_including_an_array_prompt_command(self): + # The script appends to PROMPT_COMMAND as an array; assigning the empty + # string would only clear element 0 and leave systemd's hook running. + assert _PROMPT_HOOKS_OFF.startswith(" ") + assert "unset PS0 PROMPT_COMMAND" in _PROMPT_HOOKS_OFF + + def test_the_prompt_is_read_again_once_the_hooks_are_gone(self): + # netmiko learned the prompt with the sequences in front of it; a prompt + # carrying a UUID never matches again, so strip_prompt would keep failing. + device = MagicMock() + device.send_command.return_value = "/usr/bin/apt" + + _opened(device) + + calls = [(c[0], c.args[0] if c.args else None) for c in device.mock_calls] + hooks_off = calls.index(("send_command", _PROMPT_HOOKS_OFF)) + assert "set_base_prompt" in [name for name, _ in calls[hooks_off + 1 :]] + + def test_a_shell_that_refuses_it_does_not_stop_the_connection(self): + device = MagicMock() + device.send_command.side_effect = ["", Exception("no prompt"), "", "/usr/bin/apt"] + + driver = _opened(device) + + assert driver._device is device + + +class TestSendStripsTerminalCodes: + def test_os_version_comes_back_without_the_context_sequences(self): + device = MagicMock() + device.send_command.return_value = _UBUNTU_2604_OS_RELEASE + driver = LinuxDriver("h", "admin", "p") + driver._device = device + + assert driver._send("cat /etc/os-release") == "Ubuntu 26.04.1 LTS"