Christian Manivong
51ee33eebe
CI / test (3.10) (push) Failing after 16s
CI / test (3.11) (push) Failing after 16s
CI / test (3.12) (push) Successful in 39s
CI / test (3.10) (pull_request) Successful in 38s
CI / test (3.11) (pull_request) Successful in 36s
CI / test (3.12) (pull_request) Successful in 38s
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
141 lines
3.9 KiB
Python
141 lines
3.9 KiB
Python
"""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")
|