Merge pull request 'fix: keep systemd's OSC 3008 shell context out of command output (#41)' (#42) from fix/osc-context-41 into main

CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.15.1] - 2026-10-23
+
+### Fixed
+- Ubuntu 26.04 hosts could not be polled (#41). systemd 258's
+  `/etc/profile.d/80-systemd-osc-context.sh` wraps every command of an interactive bash
+  in OSC 3008 sequences, so they ended up in every value read through the shell, the
+  prompt line with them: `os_version` grew past 255 characters. The session now unsets
+  `PS0` and `PROMPT_COMMAND` right after switching the history off and reads its prompt
+  anew, and `_send()` strips terminal codes from what it returns.
+
 ## [0.15.0] - 2026-10-23
 
 The first release on PyPI. It collects everything since 0.4.0, which was the last version recorded here.

napalm_linux/_shell.py

@@ -0,0 +1,54 @@
+"""The interactive shell netmiko opens, made fit to be read before the first command."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+#: The end of a command's output when nothing better is known: a line that looks
+#: like a shell prompt.
+PROMPT_RE = r"[#$\>]\s*$"
+
+#: Sent before anything else in the interactive session (NetOrk/netork#1039):
+#: the shell writes no history file and records nothing more. The leading space
+#: keeps this line itself out of a history under ``HISTCONTROL=ignorespace``;
+#: 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"
+
+#: Keeps the PTY from wrapping long output lines (e.g. docker JSON).
+NO_LINE_WRAP = "stty cols 10000 2>/dev/null || true"
+
+
+def prepare_shell(device: Any, hostname: str) -> None:
+    """Make netmiko's fresh shell *device* fit to read, one step at a time.
+
+    A step the shell refuses is logged and skipped: none of them is worth
+    failing the connection for.
+    """
+    # No history before anything else is typed (NetOrk/netork#1039).
+    try:
+        device.send_command(HISTORY_OFF, expect_string=PROMPT_RE)
+    except Exception:
+        logger.debug("Could not switch the shell history off on %s", hostname, exc_info=True)
+
+    # No systemd OSC 3008 context around every command (#41). netmiko learned
+    # the prompt with those sequences in front, and a prompt carrying a fresh
+    # UUID never matches again, so it reads the prompt anew once they are gone.
+    try:
+        device.send_command(PROMPT_HOOKS_OFF, expect_string=PROMPT_RE)
+        device.set_base_prompt()
+    except Exception:
+        logger.debug("Could not switch the prompt hooks off on %s", hostname, exc_info=True)
+
+    try:
+        device.send_command(NO_LINE_WRAP, expect_string=PROMPT_RE)
+    except Exception:
+        logger.debug("%s: ignored", hostname, exc_info=True)

napalm_linux/linux.py

@@ -74,6 +74,9 @@ from netmiko.exceptions import (
     NetmikoTimeoutException,
 )
 
+from napalm_linux._shell import PROMPT_RE as _PROMPT_RE
+from napalm_linux._shell import prepare_shell
+
 logger = logging.getLogger("napalm_linux")
 
 # Package managers in detection order
@@ -86,9 +89,6 @@ _RC_MARKER = "__NETORK_RC="
 _RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
 
 
-#: The end of a command's output when nothing better is known: a line that looks
-#: like a shell prompt.
-_PROMPT_RE = r"[#$\>]\s*$"
 #: A command ending in ``echo __NAME=$?`` reports its exit status on a line of its
 #: own: ``_RC_MARKER`` here, ``__APT_RC=`` and ``__SVC_RC=`` in napalm-device-types.
 _STATUS_ECHO_RE = re.compile(r"echo\s+(__[A-Z_]+=)\$\?")
@@ -159,12 +159,6 @@ _PRIVILEGED_STDIN = (
     "mktemp, cat) and have root install that file."
 )
 
-#: Sent before anything else in the interactive session (NetOrk/netork#1039):
-#: the shell writes no history file and records nothing more. The leading space
-#: keeps this line itself out of a history under ``HISTCONTROL=ignorespace``;
-#: a shell without ``set -o history`` (dash) only complains, to /dev/null.
-_HISTORY_OFF = " unset HISTFILE; set +o history 2>/dev/null; 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; "
@@ -413,21 +407,7 @@ class LinuxDriver(
             raise ConnectionException(str(exc)) from exc
         self._host_key.observe(self._device.remote_conn_pre.get_transport())
 
-        # No history before anything else is typed (NetOrk/netork#1039).
-        try:
-            self._device.send_command(_HISTORY_OFF, expect_string=_PROMPT_RE)
-        except Exception:
-            logger.debug(
-                "Could not switch the shell history off on %s", self.hostname, exc_info=True
-            )
-
-        # Prevent PTY from wrapping long output lines (e.g. docker JSON).
-        try:
-            self._device.send_command(
-                "stty cols 10000 2>/dev/null || true", expect_string=r"[#$>]\s*$"
-            )
-        except Exception:
-            logger.debug("%s: ignored", self.hostname, exc_info=True)
+        prepare_shell(self._device, self.hostname)
 
         self._pkg_manager = self._forced_pkg_manager or self._detect_pkg_manager()
         logger.debug("Connected to %s, pkg_manager=%s", self.hostname, self._pkg_manager)
@@ -455,15 +435,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.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.15.0"
+version = "0.15.1"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.10"

tests/quality_baseline.json

@@ -22,7 +22,7 @@
   "napalm_linux.linux"
  ],
  "python_file_lines": {
-  "napalm_linux/linux.py": 2307
+  "napalm_linux/linux.py": 2288
  },
  "untyped_public": []
 }

tests/test_sudo_off_the_command_line.py

@@ -21,7 +21,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from napalm_linux import LinuxDriver
-from napalm_linux.linux import _HISTORY_OFF
+from napalm_linux._shell import HISTORY_OFF as _HISTORY_OFF
 
 _PW = "s3cr3t 'q' $x"
 
@@ -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()
 

tests/test_systemd_osc_context.py

@@ -0,0 +1,79 @@
+"""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._shell import HISTORY_OFF, PROMPT_HOOKS_OFF
+from napalm_linux.linux import 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"