christianmanivong
76f85051cc
- WS-Management answers are parsed with defusedxml (new dependency, defusedxml>=0.7) with DTDs forbidden. The check for "<!DOCTYPE" in the raw bytes it replaces missed a DTD in a UTF-16 answer, which expat reads too; on plain HTTP (16992) anyone in the path can send one. An answer with a DTD in any encoding is now refused as no WS-Management answer, as a UTF-8 one already was (B314, B405). - xml.sax.saxutils only escapes the request envelopes, nothing parses with SAX (nosec B406). - python -m napalm_amt.record fetches the landing page without certificate checks, as before: it sends no credentials and records what discovery sees, and AMT's certificate is self-signed unless one was provisioned (nosec B501). The driver itself still verifies by default (verify_ssl).
317 lines
11 KiB
Python
317 lines
11 KiB
Python
"""The WS-Management client: envelopes, Digest auth, enumeration, methods and faults."""
|
|
|
|
import pytest
|
|
import requests
|
|
import responses
|
|
|
|
from napalm_amt.wsman import (
|
|
WsmanAuthError,
|
|
WsmanError,
|
|
WsmanFault,
|
|
WsmanTransport,
|
|
resource_uri,
|
|
)
|
|
|
|
from .conftest import HOST, URL, Request, load, patched
|
|
|
|
|
|
def _transport(**kwargs) -> WsmanTransport:
|
|
return WsmanTransport(HOST, "admin", "P@ssw0rd", **kwargs)
|
|
|
|
|
|
# --- resource URIs ---------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("resource", "uri"),
|
|
[
|
|
("AMT_GeneralSettings", "http://intel.com/wbem/wscim/1/amt-schema/1/AMT_GeneralSettings"),
|
|
(
|
|
"IPS_KVMRedirectionSettingData",
|
|
"http://intel.com/wbem/wscim/1/ips-schema/1/IPS_KVMRedirectionSettingData",
|
|
),
|
|
("CIM_Processor", "http://schemas.dmtf.org/wbem/wscim/1/cim-schema/2/CIM_Processor"),
|
|
],
|
|
)
|
|
def test_resource_uri_follows_the_class_prefix(resource, uri):
|
|
assert resource_uri(resource) == uri
|
|
|
|
|
|
def test_resource_uri_refuses_an_unknown_prefix():
|
|
with pytest.raises(ValueError):
|
|
resource_uri("Win32_Processor")
|
|
|
|
|
|
def test_url_is_plain_http_on_16992_and_tls_on_16993():
|
|
assert _transport().url == f"http://{HOST}:16992/wsman"
|
|
assert _transport(port=16993).url == f"https://{HOST}:16993/wsman"
|
|
assert _transport(port=8443, tls=True).url == f"https://{HOST}:8443/wsman"
|
|
|
|
|
|
# --- reading ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_returns_the_instance_as_a_dict(amt):
|
|
settings = _transport().get("AMT_GeneralSettings")
|
|
assert settings["HostName"] == "Test Host Name"
|
|
assert settings["AMTNetworkEnabled"] == "1"
|
|
request = amt.calls("Get")[0]
|
|
assert request.action == "http://schemas.xmlsoap.org/ws/2004/09/transfer/Get"
|
|
assert request.resource_uri == resource_uri("AMT_GeneralSettings")
|
|
|
|
|
|
def test_enumerate_pulls_every_instance(amt):
|
|
items = _transport().enumerate("CIM_PhysicalMemory")
|
|
assert [i["BankLabel"] for i in items] == ["BANK 0", "BANK 2"]
|
|
assert [r.operation for r in amt.requests] == ["Enumerate", "Pull"]
|
|
assert amt.calls("Pull")[0].context == "ctx-CIM_PhysicalMemory"
|
|
|
|
|
|
def test_enumerate_follows_the_context_until_end_of_sequence(amt):
|
|
# CIM_Chip's first page carries a context and no EndOfSequence: pull again.
|
|
items = _transport().enumerate("CIM_Chip")
|
|
assert [i["Version"] for i in items] == ["Intel(R) Core(TM) i7-9850H CPU @ 2.60GHz"]
|
|
assert [r.operation for r in amt.requests] == ["Enumerate", "Pull", "Pull"]
|
|
assert amt.calls("Pull")[1].context == "D4020000-0000-0000-0000-000000000000"
|
|
|
|
|
|
def test_repeated_elements_become_lists(amt):
|
|
(service,) = _transport().enumerate("CIM_AssociatedPowerManagementService")
|
|
assert service["PowerState"] == "2"
|
|
assert service["AvailableRequestedPowerStates"] == ["10", "8", "5", "11", "4", "7", "14", "12"]
|
|
|
|
|
|
def test_nested_elements_become_dicts(amt):
|
|
(bios,) = _transport().enumerate("CIM_BIOSElement")
|
|
assert bios["ReleaseDate"] == {"Datetime": "2020-08-10T00:00:00Z"}
|
|
|
|
|
|
def test_an_empty_element_is_an_empty_string(amt):
|
|
kvm = _transport().get("IPS_KVMRedirectionSettingData")
|
|
assert kvm["RFBPassword"] == ""
|
|
|
|
|
|
def test_invoke_returns_the_output_parameters(amt):
|
|
out = _transport().invoke("AMT_SetupAndConfigurationService", "GetUuid")
|
|
assert out == {"UUID": "E67jVdK/u2EXoIiu3XA36g=="}
|
|
request = amt.calls("GetUuid")[0]
|
|
assert request.action == (
|
|
"http://intel.com/wbem/wscim/1/amt-schema/1/AMT_SetupAndConfigurationService/GetUuid"
|
|
)
|
|
assert (
|
|
request.body.find(
|
|
"{http://intel.com/wbem/wscim/1/amt-schema/1/AMT_SetupAndConfigurationService}"
|
|
"GetUuid_INPUT"
|
|
)
|
|
is not None
|
|
)
|
|
|
|
|
|
def test_invoke_sends_the_input_parameters_in_order(amt):
|
|
_transport().invoke(
|
|
"AMT_MessageLog", "GetRecords", [("IterationIdentifier", "1"), ("MaxReadRecords", "390")]
|
|
)
|
|
body = amt.calls("GetRecords")[0].body
|
|
ns = "{http://intel.com/wbem/wscim/1/amt-schema/1/AMT_MessageLog}"
|
|
params = body.find(f"{ns}GetRecords_INPUT")
|
|
assert [(p.tag.removeprefix(ns), p.text) for p in params] == [
|
|
("IterationIdentifier", "1"),
|
|
("MaxReadRecords", "390"),
|
|
]
|
|
|
|
|
|
def test_invoke_accepts_raw_xml_for_references(amt):
|
|
ref = "<h:ManagedElement><Address>x</Address></h:ManagedElement>"
|
|
_transport().invoke(
|
|
"CIM_PowerManagementService", "RequestPowerStateChange", [("PowerState", "8")], raw=ref
|
|
)
|
|
xml = amt.calls("RequestPowerStateChange")[0].xml
|
|
assert "<h:PowerState>8</h:PowerState><h:ManagedElement>" in xml
|
|
|
|
|
|
def test_selectors_go_into_the_header(amt):
|
|
_transport().get("AMT_GeneralSettings", selectors={"InstanceID": "Intel(r) AMT: General"})
|
|
assert amt.calls("Get")[0].selectors == {"InstanceID": "Intel(r) AMT: General"}
|
|
|
|
|
|
def test_values_are_escaped(amt):
|
|
_transport().invoke("AMT_MessageLog", "GetRecords", [("Tag", "<a&b>")])
|
|
assert "<a&b>" in amt.calls("GetRecords")[0].xml
|
|
|
|
|
|
def test_message_ids_differ_per_request(amt):
|
|
transport = _transport()
|
|
transport.get("AMT_GeneralSettings")
|
|
transport.get("AMT_GeneralSettings")
|
|
ids = [r.xml.split("<a:MessageID>")[1].split("<")[0] for r in amt.requests]
|
|
assert len(set(ids)) == 2
|
|
|
|
|
|
# --- errors ----------------------------------------------------------------------------
|
|
|
|
|
|
def test_a_fault_raises_with_its_subcode_and_reason(amt):
|
|
with pytest.raises(WsmanFault) as excinfo:
|
|
_transport().get("AMT_NoSuchClass")
|
|
fault = excinfo.value
|
|
assert fault.subcode == "DestinationUnreachable"
|
|
assert "No route can be determined" in fault.reason
|
|
assert fault.detail.endswith("/InvalidResourceURI")
|
|
assert isinstance(fault, WsmanError)
|
|
|
|
|
|
def test_an_unknown_class_in_an_enumeration_raises_the_fault(amt):
|
|
with pytest.raises(WsmanFault):
|
|
_transport().enumerate("CIM_NoSuchClass")
|
|
|
|
|
|
def test_a_method_returning_an_error_code_raises(amt):
|
|
amt.override(
|
|
"AMT_MessageLog",
|
|
"PositionToFirstRecord",
|
|
200,
|
|
patched(
|
|
"AMT_MessageLog.PositionToFirstRecord.xml",
|
|
"<g:ReturnValue>0</g:ReturnValue>",
|
|
"<g:ReturnValue>2</g:ReturnValue>",
|
|
),
|
|
)
|
|
with pytest.raises(WsmanError, match="ReturnValue 2"):
|
|
_transport().invoke("AMT_MessageLog", "PositionToFirstRecord")
|
|
|
|
|
|
def test_a_method_may_be_allowed_an_error_code(amt):
|
|
amt.override(
|
|
"AMT_MessageLog",
|
|
"PositionToFirstRecord",
|
|
200,
|
|
patched(
|
|
"AMT_MessageLog.PositionToFirstRecord.xml",
|
|
"<g:ReturnValue>0</g:ReturnValue>",
|
|
"<g:ReturnValue>2</g:ReturnValue>",
|
|
),
|
|
)
|
|
out = _transport().invoke("AMT_MessageLog", "PositionToFirstRecord", check=False)
|
|
assert out["ReturnValue"] == "2"
|
|
|
|
|
|
def test_wrong_credentials_raise_an_auth_error():
|
|
with responses.RequestsMock() as rsps:
|
|
rsps.add(
|
|
responses.POST,
|
|
URL,
|
|
status=401,
|
|
headers={"WWW-Authenticate": 'Digest realm="Digest:0", nonce="n", qop="auth"'},
|
|
)
|
|
with pytest.raises(WsmanAuthError):
|
|
_transport(retries=0).get("AMT_GeneralSettings")
|
|
|
|
|
|
def test_a_non_soap_error_raises():
|
|
with responses.RequestsMock() as rsps:
|
|
rsps.add(responses.POST, URL, status=500, body="<html>busy</html>")
|
|
with pytest.raises(WsmanError, match="HTTP 500"):
|
|
_transport(retries=0).get("AMT_GeneralSettings")
|
|
|
|
|
|
_WITH_DTD = (
|
|
'<?xml version="1.0" encoding="{encoding}"?>'
|
|
'<!DOCTYPE Envelope [<!ENTITY a "aaaa">]>'
|
|
'<Envelope xmlns="http://www.w3.org/2003/05/soap-envelope"><Body>&a;</Body></Envelope>'
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("encoding", ["utf-8", "utf-16"])
|
|
def test_an_answer_with_a_dtd_is_refused(encoding):
|
|
"""A DTD has no business in a WS-Management answer, in whatever encoding it comes."""
|
|
with responses.RequestsMock() as rsps:
|
|
rsps.add(responses.POST, URL, body=_WITH_DTD.format(encoding=encoding).encode(encoding))
|
|
with pytest.raises(WsmanError, match="no WS-Management answer"):
|
|
_transport(retries=0).get("AMT_GeneralSettings")
|
|
|
|
|
|
def test_an_answer_that_is_no_xml_raises():
|
|
with responses.RequestsMock() as rsps:
|
|
rsps.add(responses.POST, URL, status=200, body=b"<Envelope><Body>")
|
|
with pytest.raises(WsmanError, match="HTTP 200"):
|
|
_transport(retries=0).get("AMT_GeneralSettings")
|
|
|
|
|
|
def test_a_connection_error_is_retried_then_raised():
|
|
with responses.RequestsMock() as rsps:
|
|
rsps.add(responses.POST, URL, body=requests.ConnectionError("refused"))
|
|
rsps.add(responses.POST, URL, body=load("AMT_GeneralSettings.get.xml"))
|
|
assert _transport(retries=1).get("AMT_GeneralSettings")["HostName"] == "Test Host Name"
|
|
with responses.RequestsMock() as rsps:
|
|
rsps.add(responses.POST, URL, body=requests.ConnectionError("refused"))
|
|
with pytest.raises(WsmanError, match="refused"):
|
|
_transport(retries=0).get("AMT_GeneralSettings")
|
|
|
|
|
|
# --- Digest authentication -------------------------------------------------------------
|
|
|
|
|
|
def test_digest_challenge_is_answered():
|
|
"""AMT accepts only HTTP Digest; the first request draws the challenge."""
|
|
seen = []
|
|
|
|
def handle(request):
|
|
seen.append(request.headers.get("Authorization"))
|
|
if not request.headers.get("Authorization"):
|
|
return (
|
|
401,
|
|
{
|
|
"WWW-Authenticate": 'Digest realm="Digest:F3EB554784E729164447A89F60B641C5", '
|
|
'nonce="dGVzdG5vbmNl", stale="false", qop="auth"'
|
|
},
|
|
"",
|
|
)
|
|
return 200, {}, load("AMT_GeneralSettings.get.xml")
|
|
|
|
with responses.RequestsMock() as rsps:
|
|
rsps.add_callback(responses.POST, URL, callback=handle)
|
|
assert _transport().get("AMT_GeneralSettings")["HostName"] == "Test Host Name"
|
|
assert seen[0] is None
|
|
assert seen[1].startswith("Digest ")
|
|
assert 'username="admin"' in seen[1]
|
|
assert 'uri="/wsman"' in seen[1]
|
|
assert 'qop="auth"' in seen[1] or "qop=auth" in seen[1]
|
|
|
|
|
|
def test_request_is_a_soap_post():
|
|
with responses.RequestsMock() as rsps:
|
|
rsps.add(responses.POST, URL, body=load("AMT_GeneralSettings.get.xml"))
|
|
_transport().get("AMT_GeneralSettings")
|
|
sent = rsps.calls[0].request
|
|
assert sent.headers["Content-Type"].startswith("application/soap+xml")
|
|
request = Request(sent.body)
|
|
assert request.resource == "AMT_GeneralSettings"
|
|
|
|
|
|
# --- recording -------------------------------------------------------------------------
|
|
|
|
|
|
def test_recorder_sees_every_raw_answer(amt):
|
|
seen = []
|
|
transport = _transport()
|
|
transport.recorder = lambda resource, operation, content: seen.append(
|
|
(resource, operation, content[:5])
|
|
)
|
|
transport.enumerate("CIM_PhysicalMemory")
|
|
transport.invoke("AMT_SetupAndConfigurationService", "GetUuid")
|
|
assert [(r, o) for r, o, _ in seen] == [
|
|
("CIM_PhysicalMemory", "Enumerate"),
|
|
("CIM_PhysicalMemory", "Pull"),
|
|
("AMT_SetupAndConfigurationService", "GetUuid"),
|
|
]
|
|
assert all(c == b"<?xml" for _, _, c in seen)
|
|
|
|
|
|
def test_recorder_sees_faults_too(amt):
|
|
seen = []
|
|
transport = _transport()
|
|
transport.recorder = lambda resource, operation, content: seen.append(operation)
|
|
with pytest.raises(WsmanFault):
|
|
transport.get("AMT_NoSuchClass")
|
|
assert seen == ["Get"]
|