Merge pull request 'feat: run commands and streams on an exec channel, reach the container engine' (#18) from feat/container-engine-channel into master

netOrk drove this driver's shell through the private `_send` (an
interactive PTY, stdout and stderr merged, no exit code) and opened its
own paramiko connections for Docker. napalm-device-types 2.6.0 makes the
channel public; this implements it (NetOrk/netork#765):

- `run_command()` and `open_stream()` open an exec channel on the SSH
  transport netmiko already holds: no second login, no PTY, a real exit
  status.
- `privileged=True` follows the driver's existing rules: as root the
  command runs directly; with a sudo password it goes through
  `sudo -S -p ''` and the password is written to stdin, never onto the
  command line; without one `sudo -n` fails at once instead of hanging.
- `ContainerEngineMixin` is mixed in, so `open_container_engine("docker")`
  streams the Engine API over `docker system dial-stdio`.

The existing Docker methods are unchanged. Version 0.2.0, requires
napalm-device-types >= 2.6.0.

Refs NAPALM/napalm-device-types#17

.github/workflows/ci.yml

@@ -0,0 +1,48 @@
+name: CI
+
+on:
+  push:
+    branches: ["**"]
+  pull_request:
+    branches: ["**"]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12"]
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+
+      - name: Install package with dev extras
+        run: |
+          python -m pip install --upgrade pip
+          # napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
+          # pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
+          python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
+          python -m pip install -e ".[dev]"
+
+      - name: Run unit tests
+        run: |
+          python -m pytest -q --tb=short
+
+      - name: Build wheel and sdist
+        run: |
+          python -m pip install build
+          python -m build
+
+      - name: Upload dist artifacts
+        # v4 refuses to run on Gitea ("not currently supported on GHES").
+        uses: actions/upload-artifact@v3
+        with:
+          name: dist-${{ matrix.python-version }}
+          path: dist/*

CHANGELOG.md

@@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.2.0] – 2026-10-07
+
+### Added
+
+- `run_command()` and `open_stream()`, the public command channel from
+  napalm-device-types 2.6.0: an exec channel on the existing SSH transport, so
+  no PTY, separate stderr and a real exit code. `privileged=True` runs as root
+  directly, through `sudo -S` with the password on stdin (never on the command
+  line), or through `sudo -n`, which fails instead of prompting.
+- `ContainerEngineMixin`: `container_engines()` and `open_container_engine()`,
+  whose `open_api()` streams the Docker Engine API over `docker system
+  dial-stdio` (NetOrk/netork#765). The existing Docker methods are unchanged.
+
+### Changed
+
+- Requires napalm-device-types >= 2.6.0.
+
 ## [0.1.0] – 2026-05-29
 
 ### Added

napalm_linux/linux.py

@@ -34,14 +34,20 @@ from napalm.base.netmiko_helpers import netmiko_args
 from napalm_device_types import (
     APT_UPGRADABLE_COMMAND,
     DNF_SECURITY_COMMAND,
+    ByteStream,
+    CommandResult,
+    ContainerEngineMixin,
     FingerprintRule,
     HostStatusMixin,
     KernelFactsMixin,
+    ListeningSocketsMixin,
     OSDriver,
     SystemdServicesMixin,
     SystemdUnavailable,
     parse_apt_upgradable,
+    open_stream_on_transport,
     parse_dnf_security,
+    run_on_transport,
     strip_terminal_codes,
 )
 from napalm_device_types.models import (
@@ -69,6 +75,31 @@ _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_]+=)\$\?")
+
+
+def _expect_for(command: str) -> str:
+    """The pattern that ends *command*'s output.
+
+    netmiko stops reading as soon as the pattern matches what it has read so far.
+    A line the command prints can end in ``#``, ``$`` or ``>`` -- apt's
+    ``<ftpmaster@ubuntu.com>`` after a bad signature -- and was taken for the
+    prompt: half the output came back, and the rest started the next command's
+    (#615). A command that echoes its exit status is read until that marker, with
+    a number, and the prompt line after it. The echoed command line carries a
+    literal ``$?`` and cannot match.
+    """
+    markers = _STATUS_ECHO_RE.findall(command)
+    if not markers:
+        return _PROMPT_RE
+    return re.escape(markers[-1]) + r"\d+\s*\n.*" + _PROMPT_RE
+
+
 def _split_status(raw: str) -> tuple[str, int | None]:
     """``(output, exit_status)`` of a command followed by ``echo {_RC_MARKER}$?``.
 
@@ -95,6 +126,11 @@ _REFRESH = {
 }
 _YUM_SECURITY_COMMAND = "LC_ALL=C yum updateinfo list security -q 2>/dev/null"
 
+#: Restart the host two seconds later, detached from this session: the launcher's
+#: exit status comes back before the host goes down, and closing the session
+#: cannot take the restart with it.
+_REBOOT_DETACHED = "sh -c '(trap \"\" HUP; sleep 2; /sbin/reboot) </dev/null >/dev/null 2>&1 &'"
+
 #: What to do when sudo wants a password netOrk does not have.
 _SUDO_PASSWORD_HINT = (
     "sudo requires a password on this device but none is configured in netOrk. "
@@ -184,7 +220,14 @@ def _short_image_id(raw: str) -> str:
     return raw.strip().removeprefix("sha256:")[:12]
 
 
-class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDriver):
+class LinuxDriver(
+    KernelFactsMixin,
+    ListeningSocketsMixin,
+    SystemdServicesMixin,
+    HostStatusMixin,
+    ContainerEngineMixin,
+    OSDriver,
+):
     """NAPALM driver for generic Linux systems.
 
     Connects via SSH (netmiko ``linux`` device type) and auto-detects the
@@ -306,7 +349,7 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDri
             command,
             read_timeout=read_timeout,
             cmd_verify=False,
-            expect_string=r'[#$\>]\s*$',
+            expect_string=_expect_for(command),
         ).strip()
 
     def _sudo(self, command: str, read_timeout: float = 100) -> str:
@@ -364,6 +407,65 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDri
             return self._sudo(command, read_timeout=timeout)
         return self._send(f"sudo -n {command}", read_timeout=timeout)
 
+    # ------------------------------------------------------------------
+    # Command channel (napalm-device-types CommandChannelMixin)
+    # ------------------------------------------------------------------
+
+    def _transport(self) -> Any:
+        """The SSH transport netmiko already holds, for exec channels next to its PTY."""
+        if not self._device:
+            raise ConnectionClosedException("Not connected")
+        return self._device.remote_conn_pre.get_transport()
+
+    def _privileged(self, command: str, privileged: bool) -> tuple[str, bytes | None]:
+        """The command line that runs *command* with the privileges asked for,
+        and what has to reach its stdin first.
+
+        Root runs it directly. With a sudo password, ``sudo -S`` reads it from
+        stdin, so it never appears in a process list. Without one, ``sudo -n``
+        fails at once where a prompt would hang. The command goes to ``sh -c``
+        as one argument, so the privilege covers every part of it.
+        """
+        if not privileged or self._is_root():
+            return command, None
+        if self._sudo_password:
+            line = f"sudo -S -p '' sh -c {_shlex_quote(command)}"
+            return line, f"{self._sudo_password}\n".encode()
+        return f"sudo -n sh -c {_shlex_quote(command)}", None
+
+    def run_command(
+        self,
+        command: str,
+        *,
+        privileged: bool = False,
+        timeout: float = 60,
+        stdin: bytes | None = None,
+    ) -> CommandResult:
+        """Run *command* on an exec channel: no PTY, stderr apart, a real exit code."""
+        line, prefix = self._privileged(command, privileged)
+        data = (prefix or b"") + (stdin or b"") if (prefix or stdin) else None
+        return run_on_transport(self._transport(), line, stdin=data, timeout=timeout)
+
+    def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
+        """Start *command* on an exec channel and return a stream to it."""
+        line, prefix = self._privileged(command, privileged)
+        return open_stream_on_transport(self._transport(), line, stdin_prefix=prefix)
+
+    def reboot_host(self) -> None:
+        """Restart the host (``HostRebootMixin``); returns once the restart is under way.
+
+        :raises RuntimeError: when the host refuses -- sudo without a password,
+            no ``reboot`` -- or its answer carried no exit status.
+        """
+        output, status = _split_status(
+            self._run_privileged(f"{_REBOOT_DETACHED}; echo {_RC_MARKER}$?", 30)
+        )
+        if status != 0:
+            reason = output or f"the reboot command exited with status {status}"
+            if "password is required" in output:
+                reason = f"{reason}\n{_SUDO_PASSWORD_HINT}"
+            raise RuntimeError(reason)
+
     def _run_host_status_command(self, command: str) -> str:
         """The transport for ``HostStatusMixin.get_host_status``: read-only, no sudo."""
         return self._send(command, read_timeout=60)
@@ -924,6 +1026,19 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDri
         """The transport for ``KernelFactsMixin.get_kernel_facts``: read-only, no sudo."""
         return self._send(command, read_timeout=60)
 
+    # ------------------------------------------------------------------
+    # ListeningSocketsMixin – the transport for get_listening_sockets
+    # ------------------------------------------------------------------
+
+    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``ListeningSocketsMixin.get_listening_sockets``.
+
+        Read-only either way; root only so that ``ss`` names every process.
+        """
+        if privileged:
+            return self._run_privileged(command, 60)
+        return self._send(command, read_timeout=60)
+
     # ------------------------------------------------------------------
     # OSDriver – package management
     # ------------------------------------------------------------------

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.1.0"
+version = "0.2.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=2.3.0",
+    "napalm-device-types>=2.6.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_command_channel.py

@@ -0,0 +1,140 @@
+"""The public command channel and container engine access (napalm-device-types 2.6.0).
+
+netOrk used to reach a Linux host's shell through the private ``_send``: an
+interactive PTY, stdout and stderr merged, no exit code. ``run_command`` and
+``open_stream`` go through an exec channel on the same SSH transport instead,
+and ``open_container_engine`` builds on them (NetOrk/netork#765). Privileges
+work as they do everywhere else in this driver: root runs directly, a sudo
+password goes to ``sudo -S`` on stdin and never onto a command line, and
+without one ``sudo -n`` fails at once instead of hanging.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+from napalm.base.exceptions import ConnectionClosedException
+
+from napalm_linux import LinuxDriver
+
+
+class FakeChannel:
+    def __init__(self):
+        self.command = None
+        self.sent = b""
+
+    def exec_command(self, command):
+        self.command = command
+
+    def settimeout(self, timeout):
+        pass
+
+    def sendall(self, data):
+        self.sent += data
+
+    def shutdown_write(self):
+        pass
+
+    def close(self):
+        pass
+
+    def recv_ready(self):
+        return False
+
+    def recv_stderr_ready(self):
+        return False
+
+    def exit_status_ready(self):
+        return True
+
+    def recv_exit_status(self):
+        return 0
+
+
+class FakeTransport:
+    def __init__(self):
+        self.channels = []
+
+    def open_session(self):
+        self.channels.append(FakeChannel())
+        return self.channels[-1]
+
+
+def _driver(*, root=False, sudo_password=None):
+    driver = LinuxDriver("h", "u", "p", optional_args={"sudo_password": sudo_password})
+    transport = FakeTransport()
+    driver._device = SimpleNamespace(remote_conn_pre=SimpleNamespace(get_transport=lambda: transport))
+    driver._root = root
+    return driver, transport
+
+
+def test_an_unprivileged_command_runs_as_given():
+    driver, transport = _driver()
+
+    result = driver.run_command("docker version", timeout=5)
+
+    assert transport.channels[-1].command == "docker version"
+    assert transport.channels[-1].sent == b""
+    assert result.exit_code == 0
+
+
+def test_a_privileged_command_with_a_sudo_password_reads_it_from_stdin():
+    driver, transport = _driver(sudo_password="s3cr3t")
+
+    driver.run_command("usermod -aG docker u", privileged=True, timeout=5)
+
+    channel = transport.channels[-1]
+    assert channel.command == "sudo -S -p '' sh -c 'usermod -aG docker u'"
+    assert channel.sent == b"s3cr3t\n"
+    assert "s3cr3t" not in channel.command
+
+
+def test_a_privileged_command_without_a_password_fails_fast_instead_of_prompting():
+    driver, transport = _driver()
+
+    driver.run_command("id", privileged=True, timeout=5)
+
+    assert transport.channels[-1].command == "sudo -n sh -c id"
+
+
+def test_a_root_login_needs_no_sudo():
+    driver, transport = _driver(root=True, sudo_password="s3cr3t")
+
+    driver.run_command("id", privileged=True, timeout=5)
+
+    assert transport.channels[-1].command == "id"
+    assert transport.channels[-1].sent == b""
+
+
+def test_stdin_follows_the_sudo_password():
+    driver, transport = _driver(sudo_password="pw")
+
+    driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5)
+
+    assert transport.channels[-1].sent == b"pw\ndata"
+
+
+def test_a_privileged_stream_gets_the_password_first():
+    driver, transport = _driver(sudo_password="pw")
+
+    driver.open_stream("cat > /tmp/x", privileged=True)
+
+    channel = transport.channels[-1]
+    assert channel.command == "sudo -S -p '' sh -c 'cat > /tmp/x'"
+    assert channel.sent == b"pw\n"
+
+
+def test_the_container_engine_api_is_a_stream_over_dial_stdio():
+    driver, transport = _driver()
+
+    driver.open_container_engine("docker").open_api()
+
+    assert transport.channels[-1].command == "docker system dial-stdio"
+
+
+def test_the_channel_needs_an_open_connection():
+    driver = LinuxDriver("h", "u", "p")
+
+    with pytest.raises(ConnectionClosedException):
+        driver.run_command("true")

tests/test_linux.py

@@ -1,5 +1,7 @@
 """Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed)."""
 
+import re
+
 import pytest
 from unittest.mock import MagicMock, patch
 from napalm_linux.linux import LinuxDriver, _arm_vendor_from_model
@@ -1050,6 +1052,106 @@ class TestSudoStatus:
         assert "__NETORK_RC=1" not in output
 
 
+class _Channel:
+    """A netmiko connection that hands out its output in chunks and stops where
+    netmiko does: at the first chunk after which ``expect_string`` matches all
+    that was read so far."""
+
+    def __init__(self, chunks):
+        self.chunks = list(chunks)
+        self.patterns: list = []
+
+    def send_command(self, command, *, expect_string, **_kwargs):
+        self.patterns.append(expect_string)
+        output = ""
+        while self.chunks:
+            output += self.chunks.pop(0)
+            if re.search(expect_string, output):
+                return output
+        raise TimeoutError(f"pattern not detected: {expect_string!r}")
+
+
+#: What vault-01 sent on 2026-10-06 while its apt proxy served a corrupted
+#: InRelease: the signature line ends in ">", which looks like a prompt (#615).
+_BADSIG_CHUNKS = [
+    "sudo -n apt-get update -q 2>&1; echo __NETORK_RC=$?\n",
+    "Fehl:2 http://archive.ubuntu.com/ubuntu noble-updates InRelease\n"
+    "  Die folgenden Signaturen waren ungültig: BADSIG 871920D1991BC93C "
+    "Ubuntu Archive Automatic Signing Key (2018) <ftpmaster@ubuntu.com>\n",
+    "W: Fehler beim Holen von http://archive.ubuntu.com/ubuntu/dists/noble-updates/InRelease\n"
+    "E: Das Depot ist nicht signiert.\n__NETORK_RC=100\n",
+    "chris@vault-01:~$ ",
+]
+
+
+class TestReadToTheEnd:
+    """A line the command prints can end in ``>``, ``#`` or ``$`` -- apt's
+    ``<ftpmaster@ubuntu.com>`` after a bad signature. Taken for the prompt, it
+    ended the read while the command still ran, and the rest arrived as the next
+    command's output (#615). A command that echoes its exit status is read until
+    that marker and the prompt after it."""
+
+    def test_a_signature_line_does_not_end_the_refresh(self, driver):
+        driver._root = False
+        driver._device = _Channel(_BADSIG_CHUNKS)
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is False
+        assert "E: Das Depot ist nicht signiert." in result["output"]
+
+    def test_the_session_stays_in_step(self, driver):
+        """Everything up to the prompt is consumed, so the next command reads its own output."""
+        driver._root = False
+        driver._device = _Channel(_BADSIG_CHUNKS + ["true\n", "__NETORK_RC=0\nchris@vault-01:~$ "])
+
+        driver.refresh_available_updates()
+        output, status = driver._sudo_status("true")
+
+        assert status == 0
+        assert "BADSIG" not in output
+
+    def test_the_echoed_command_does_not_count_as_the_marker(self, driver):
+        """Its literal ``$?`` is no number."""
+        channel = _Channel(["sudo true; echo __NETORK_RC=$?\n", "__NETORK_RC=0\nchris@vault-01:~$ "])
+        driver._device = channel
+
+        assert driver._sudo_status("true")[1] == 0
+        assert channel.chunks == []
+
+    def test_the_marker_alone_is_not_the_end(self, driver):
+        """The prompt after it has to be read too, or it would start the next output."""
+        channel = _Channel(["out\n__NETORK_RC=0\n", "chris@vault-01:~$ "])
+        driver._device = channel
+
+        driver._sudo_status("true")
+
+        assert channel.chunks == []
+
+    @pytest.mark.parametrize(
+        "command",
+        [
+            "{ LC_ALL=C apt list --upgradable 2>/dev/null; echo __APT_RC=$?; } | cat",
+            "timeout 45 systemctl restart -- cron.service; echo __SVC_RC=$?",
+        ],
+    )
+    def test_every_status_marker_is_waited_for(self, driver, command):
+        marker = re.search(r"echo (__[A-Z_]+=)", command).group(1)
+        channel = _Channel([f"x <a@b>\n", f"{marker}0\nchris@host:~$ "])
+        driver._device = channel
+
+        output = driver._send(command)
+
+        assert f"{marker}0" in output
+
+    def test_a_command_without_a_marker_still_ends_at_the_prompt(self, driver):
+        channel = _Channel(["6.8.0-142-generic\nchris@host:~$ "])
+        driver._device = channel
+
+        assert driver._send("uname -r").startswith("6.8.0-142-generic")
+        assert channel.patterns == [r"[#$\>]\s*$"]
+
+
 class TestUninstallExitStatus:
     """Whether a removal worked is what the package manager's exit status says.
 
@@ -1415,3 +1517,103 @@ class TestTerminalCodes:
         driver._device.send_command.side_effect = ["0", "Hit:1 noble InRelease\n\x1b>__NETORK_RC=0"]
 
         assert driver.refresh_available_updates()["success"] is True
+
+
+class TestRebootHost:
+    """``reboot_host`` (napalm-device-types' ``HostRebootMixin``) restarts the host.
+
+    Without it netOrk could not restart a Linux host at all: its capability check
+    looks for ``reboot_host`` and found nothing (netOrk #637). The restart is
+    detached and a moment late, so the launcher's exit status comes back before
+    the host goes down, and closing the session cannot take it along.
+    """
+
+    def test_the_driver_can_restart_its_host(self):
+        assert callable(getattr(LinuxDriver, "reboot_host", None))
+
+    def test_the_restart_is_detached_and_privileged(self, driver):
+        driver._root = False
+        driver._device.send_command.return_value = "\n__NETORK_RC=0"
+
+        driver.reboot_host()
+
+        sent = driver._device.send_command.call_args[0][0]
+        assert sent.startswith("sudo -n sh -c ")
+        assert "/sbin/reboot" in sent and "trap" in sent and "&" in sent
+        assert sent.endswith("echo __NETORK_RC=$?")
+
+    def test_a_refusal_is_raised_with_what_the_host_said(self, driver):
+        driver._root = False
+        driver._device.send_command.return_value = "sudo: a password is required\n__NETORK_RC=1"
+
+        with pytest.raises(RuntimeError, match="password is required"):
+            driver.reboot_host()
+
+    def test_no_exit_status_is_no_success(self, driver):
+        driver._root = True
+        driver._device.send_command.return_value = "something else"
+
+        with pytest.raises(RuntimeError):
+            driver.reboot_host()
+
+
+# ---------------------------------------------------------------------------
+# Listening sockets: ss and the processes' cgroups, as root when it can
+# ---------------------------------------------------------------------------
+
+_SOCKETS = (
+    "SOCK_BEGIN\n[ss]\n"
+    'tcp LISTEN 0 128 0.0.0.0:5432 0.0.0.0:* users:(("postgres",pid=812,fd=6))\n'
+    "__SS_RC=0\n[cgroups]\n"
+    "812 0::/system.slice/system-postgresql.slice/postgresql@16-main.service\n"
+    "SOCK_END\n"
+)
+
+
+class TestGetListeningSockets:
+    def _sent(self, driver) -> list[str]:
+        return [c[0][0] for c in driver._device.send_command.call_args_list]
+
+    def test_it_reads_with_the_shared_command(self, driver):
+        from napalm_device_types import ListeningSocketsMixin
+
+        assert isinstance(driver, ListeningSocketsMixin)
+        driver._root = True
+        driver._device.send_command.return_value = _SOCKETS
+
+        reading = driver.get_listening_sockets()
+
+        assert reading["attributed"] is True
+        [socket] = reading["sockets"]
+        assert (socket["port"], socket["unit"]) == (5432, "postgresql@16-main")
+        assert self._sent(driver)[0].startswith("sh -c '")
+
+    def test_without_a_sudo_password_the_whole_script_runs_under_sudo_n(self, driver):
+        driver._root = False
+        driver._device.send_command.return_value = _SOCKETS
+
+        driver.get_listening_sockets()
+
+        assert self._sent(driver)[0].startswith("sudo -n sh -c '")
+
+    def test_with_a_sudo_password_it_goes_through_sudo(self, driver):
+        driver._root = False
+        driver._sudo_password = "pw"  # noqa: S105
+        driver._device.send_command.return_value = _SOCKETS
+
+        driver.get_listening_sockets()
+
+        sent = self._sent(driver)[0]
+        assert sent.startswith("echo pw | sudo -S")
+        assert "sh -c '" in sent
+
+    def test_when_sudo_refuses_it_reads_what_the_user_may_see(self, driver):
+        driver._root = False
+        driver._device.send_command.side_effect = ["sudo: a password is required", _SOCKETS]
+
+        reading = driver.get_listening_sockets()
+
+        assert reading["attributed"] is False
+        refused, plain = self._sent(driver)
+        assert refused.startswith("sudo -n sh -c '")
+        assert plain.startswith("sh -c '")