Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c00deef3d7 |
No files matched your search
+19
-3
@@ -167,6 +167,12 @@ _PRIVILEGED_STDIN = (
|
|||||||
#: a shell without ``set -o history`` (dash) only complains, to /dev/null.
|
#: a shell without ``set -o history`` (dash) only complains, to /dev/null.
|
||||||
_HISTORY_OFF = " unset HISTFILE; set +o history 2>/dev/null; true"
|
_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,
|
#: 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.
|
#: a cached timestamp) leaves the password on stdin, and nothing may read it.
|
||||||
_NO_STDIN = "exec </dev/null; "
|
_NO_STDIN = "exec </dev/null; "
|
||||||
@@ -398,6 +404,15 @@ class LinuxDriver(
|
|||||||
except Exception:
|
except Exception:
|
||||||
logger.debug("Could not switch the shell history off on %s", self.hostname, exc_info=True)
|
logger.debug("Could not switch the shell history off on %s", self.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:
|
||||||
|
self._device.send_command(_PROMPT_HOOKS_OFF, expect_string=_PROMPT_RE)
|
||||||
|
self._device.set_base_prompt()
|
||||||
|
except Exception:
|
||||||
|
logger.debug("Could not switch the prompt hooks off on %s", self.hostname, exc_info=True)
|
||||||
|
|
||||||
# Prevent PTY from wrapping long output lines (e.g. docker JSON).
|
# Prevent PTY from wrapping long output lines (e.g. docker JSON).
|
||||||
try:
|
try:
|
||||||
self._device.send_command("stty cols 10000 2>/dev/null || true", expect_string=r"[#$>]\s*$")
|
self._device.send_command("stty cols 10000 2>/dev/null || true", expect_string=r"[#$>]\s*$")
|
||||||
@@ -430,15 +445,16 @@ class LinuxDriver(
|
|||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
|
|
||||||
def _send(self, command: str, read_timeout: float = 100) -> str:
|
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:
|
if not self._device:
|
||||||
raise ConnectionClosedException("Not connected")
|
raise ConnectionClosedException("Not connected")
|
||||||
return self._device.send_command(
|
output = self._device.send_command(
|
||||||
command,
|
command,
|
||||||
read_timeout=read_timeout,
|
read_timeout=read_timeout,
|
||||||
cmd_verify=False,
|
cmd_verify=False,
|
||||||
expect_string=_expect_for(command),
|
expect_string=_expect_for(command),
|
||||||
).strip()
|
)
|
||||||
|
return strip_terminal_codes(output).strip()
|
||||||
|
|
||||||
def _sudo(self, command: str, read_timeout: float = 100) -> str:
|
def _sudo(self, command: str, read_timeout: float = 100) -> str:
|
||||||
"""Run *command* as root and return its output, stdout and stderr together.
|
"""Run *command* as root and return its output, stdout and stderr together.
|
||||||
|
|||||||
@@ -236,7 +236,8 @@ class TestHistoryOff:
|
|||||||
def test_a_shell_that_refuses_it_does_not_stop_the_connection(self):
|
def test_a_shell_that_refuses_it_does_not_stop_the_connection(self):
|
||||||
driver = LinuxDriver("h", "admin", "p")
|
driver = LinuxDriver("h", "admin", "p")
|
||||||
device = MagicMock()
|
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):
|
with patch("napalm_linux.linux.ConnectHandler", return_value=device):
|
||||||
driver.open()
|
driver.open()
|
||||||
|
|
||||||
|
|||||||
@@ -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"
|
||||||
Reference in new issue
Block a user