napalm-linux/tests/test_command_channel.py

"""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(self, size):
        return b""

    def recv_stderr_ready(self):
        return False

    def recv_stderr(self, size):
        return b""

    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_a_privileged_command_with_a_sudo_password_takes_no_stdin():
    """sudo -S reads the password from stdin. A sudo that does not prompt --
    NOPASSWD, a cached timestamp -- leaves it there, and the command reads it as
    its first line: it wrote the password into an sshd drop-in (NetOrk/netork#1029)."""
    driver, transport = _driver(sudo_password="pw")

    with pytest.raises(ValueError, match="stdin"):
        driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5)

    assert transport.channels == []  # nothing reached the host


def test_privileged_stdin_without_a_sudo_password_goes_to_the_command():
    """sudo -n reads nothing from stdin: the data is the command's alone."""
    driver, transport = _driver()

    driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5)

    channel = transport.channels[-1]
    assert channel.command == "sudo -n sh -c 'tee /etc/x'"
    assert channel.sent == b"data"


def test_privileged_stdin_for_a_root_login_goes_to_the_command():
    driver, transport = _driver(root=True, sudo_password="pw")

    driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5)

    assert transport.channels[-1].command == "tee /etc/x"
    assert transport.channels[-1].sent == b"data"


def test_unprivileged_stdin_with_a_sudo_password_goes_to_the_command():
    driver, transport = _driver(sudo_password="pw")

    driver.run_command("cat > /tmp/x", stdin=b"data", timeout=5)

    assert transport.channels[-1].command == "cat > /tmp/x"
    assert transport.channels[-1].sent == b"data"


def test_a_privileged_stream_gets_the_password_first():
    driver, transport = _driver(sudo_password="pw")

    driver.open_stream("journalctl -f", privileged=True)

    channel = transport.channels[-1]
    assert channel.command == "sudo -S -p '' sh -c 'journalctl -f'"
    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")