"""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 = "
x
" _transport().invoke( "CIM_PowerManagementService", "RequestPowerStateChange", [("PowerState", "8")], raw=ref ) xml = amt.calls("RequestPowerStateChange")[0].xml assert "8" 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", "")]) 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("")[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", "0", "2", ), ) 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", "0", "2", ), ) 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="busy") with pytest.raises(WsmanError, match="HTTP 500"): _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"