Christian Manivong
1ce42ef099
Talks PSRP over WinRM via pypsrp, one runspace pool per connection. Each getter sends one PowerShell script that projects cmdlet results onto flat fields and ends in ConvertTo-Json, so the Python side parses JSON, not text. Covers facts, interfaces, IP addresses, ARP, routes and services, plus service start/stop/restart/enable/disable. Service names are validated and quoted as PowerShell verbatim strings, typographic quotes included. Fixtures are synthetic: they pin down the JSON the scripts are designed to emit. tools/harvest.py records the real output from a host. Refs christianmanivong/netork#300
148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
"""Unit tests for the PSRP transport.
|
|
|
|
pypsrp is patched out at the module boundary: these tests pin down how the
|
|
transport configures it and how its failures reach the driver, not pypsrp itself.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import requests
|
|
from napalm.base.exceptions import ConnectionClosedException, ConnectionException
|
|
from pypsrp.exceptions import AuthenticationError, WinRMTransportError
|
|
|
|
from napalm_windows.transport import PowerShellError, PsrpTransport
|
|
|
|
|
|
def _transport(**overrides):
|
|
kwargs = {
|
|
"port": 5986,
|
|
"ssl": True,
|
|
"cert_validation": True,
|
|
"auth": "negotiate",
|
|
"timeout": 60,
|
|
}
|
|
kwargs.update(overrides)
|
|
return PsrpTransport("win01", "admin", "secret", **kwargs)
|
|
|
|
|
|
@pytest.fixture()
|
|
def psrp():
|
|
"""Patch WSMan, RunspacePool and PowerShell as the transport sees them."""
|
|
with (
|
|
patch("napalm_windows.transport.WSMan") as wsman,
|
|
patch("napalm_windows.transport.RunspacePool") as pool,
|
|
patch("napalm_windows.transport.PowerShell") as ps,
|
|
):
|
|
yield {"wsman": wsman, "pool": pool, "ps": ps}
|
|
|
|
|
|
class TestOpen:
|
|
def test_passes_connection_settings_to_wsman(self, psrp):
|
|
_transport(port=5985, ssl=False, cert_validation=False, auth="ntlm").open()
|
|
|
|
kwargs = psrp["wsman"].call_args.kwargs
|
|
assert psrp["wsman"].call_args.args == ("win01",)
|
|
assert kwargs["port"] == 5985
|
|
assert kwargs["ssl"] is False
|
|
assert kwargs["cert_validation"] is False
|
|
assert kwargs["auth"] == "ntlm"
|
|
assert kwargs["username"] == "admin"
|
|
assert kwargs["password"] == "secret" # noqa: S105
|
|
|
|
def test_opens_one_runspace_pool_on_the_connection(self, psrp):
|
|
_transport().open()
|
|
|
|
psrp["pool"].assert_called_once_with(psrp["wsman"].return_value)
|
|
psrp["pool"].return_value.open.assert_called_once_with()
|
|
|
|
def test_authentication_failure_is_a_connection_exception(self, psrp):
|
|
psrp["pool"].return_value.open.side_effect = AuthenticationError("bad creds")
|
|
|
|
with pytest.raises(ConnectionException, match="Authentication failed"):
|
|
_transport().open()
|
|
|
|
def test_http_401_is_an_authentication_failure(self, psrp):
|
|
psrp["pool"].return_value.open.side_effect = WinRMTransportError(
|
|
"http", 401, "Unauthorized"
|
|
)
|
|
|
|
with pytest.raises(ConnectionException, match="Authentication failed"):
|
|
_transport().open()
|
|
|
|
def test_unreachable_host_is_a_connection_exception(self, psrp):
|
|
psrp["pool"].return_value.open.side_effect = requests.ConnectionError("refused")
|
|
|
|
with pytest.raises(ConnectionException, match="win01:5986"):
|
|
_transport().open()
|
|
|
|
def test_is_open_only_after_open(self, psrp):
|
|
t = _transport()
|
|
assert t.is_open is False
|
|
t.open()
|
|
assert t.is_open is True
|
|
|
|
|
|
class TestRun:
|
|
def test_returns_output_objects_joined_by_newline(self, psrp):
|
|
psrp["ps"].return_value.invoke.return_value = ['{"a":', "1}"]
|
|
psrp["ps"].return_value.had_errors = False
|
|
t = _transport()
|
|
t.open()
|
|
|
|
assert t.run("Get-Thing") == '{"a":\n1}'
|
|
psrp["ps"].return_value.add_script.assert_called_once_with("Get-Thing")
|
|
|
|
def test_none_objects_are_skipped(self, psrp):
|
|
psrp["ps"].return_value.invoke.return_value = [None, "x", None]
|
|
psrp["ps"].return_value.had_errors = False
|
|
t = _transport()
|
|
t.open()
|
|
|
|
assert t.run("Get-Thing") == "x"
|
|
|
|
def test_error_stream_raises_powershell_error_with_its_text(self, psrp):
|
|
psrp["ps"].return_value.invoke.return_value = []
|
|
psrp["ps"].return_value.had_errors = True
|
|
psrp["ps"].return_value.streams.error = [
|
|
"Cannot find any service with service name 'nope'."
|
|
]
|
|
t = _transport()
|
|
t.open()
|
|
|
|
with pytest.raises(PowerShellError, match="service name 'nope'"):
|
|
t.run("Start-Service nope")
|
|
|
|
def test_run_before_open_raises_connection_closed(self):
|
|
with pytest.raises(ConnectionClosedException):
|
|
_transport().run("Get-Thing")
|
|
|
|
|
|
class TestClose:
|
|
def test_close_closes_pool_and_connection(self, psrp):
|
|
t = _transport()
|
|
t.open()
|
|
t.close()
|
|
|
|
psrp["pool"].return_value.close.assert_called_once_with()
|
|
psrp["wsman"].return_value.close.assert_called_once_with()
|
|
assert t.is_open is False
|
|
|
|
def test_close_swallows_errors_from_a_dead_connection(self, psrp):
|
|
psrp["pool"].return_value.close.side_effect = requests.ConnectionError("gone")
|
|
t = _transport()
|
|
t.open()
|
|
|
|
t.close()
|
|
|
|
assert t.is_open is False
|
|
|
|
def test_close_without_open_is_a_no_op(self):
|
|
_transport().close()
|
|
|
|
|
|
def test_transport_does_not_print_password_in_repr():
|
|
assert "secret" not in repr(_transport())
|