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

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,6 +34,9 @@ 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,
@@ -42,7 +45,9 @@ from napalm_device_types import (
     SystemdServicesMixin,
     SystemdUnavailable,
     parse_apt_upgradable,
+    open_stream_on_transport,
     parse_dnf_security,
+    run_on_transport,
     strip_terminal_codes,
 )
 from napalm_device_types.models import (
@@ -216,7 +221,12 @@ def _short_image_id(raw: str) -> str:
 
 
 class LinuxDriver(
-    KernelFactsMixin, ListeningSocketsMixin, SystemdServicesMixin, HostStatusMixin, OSDriver
+    KernelFactsMixin,
+    ListeningSocketsMixin,
+    SystemdServicesMixin,
+    HostStatusMixin,
+    ContainerEngineMixin,
+    OSDriver,
 ):
     """NAPALM driver for generic Linux systems.
 
@@ -397,6 +407,50 @@ class LinuxDriver(
             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.
 

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.4.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")