christianmanivong
0052ba7884
An `amt` driver (napalm_amt.AmtDriver), a ManagementControllerDriver for the management controller of Intel vPro machines, with its own lean WS-Management client (requests + ElementTree, HTTP Digest, ports 16992/16993). It reads the machine's identity, hardware with the SMBIOS system UUID, firmware (AMT with build, BIOS), power state and the hardware event log, reports what AMT listens on (plain HTTP or TLS, KVM incl. port 5900, IDE-R), and switches the machine's power. AMT cannot be reset over WS-Management, so there is no reboot_host. Discovery recognises AMT by its ports and its Server header. Tested against Intel's published AMT 12 recordings (go-wsman-messages, Apache-2.0, see NOTICE). `python -m napalm_amt.record` records a real AMT read-only; the first test devices are two Lenovo ThinkCentre M910x.
119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
"""Recording fixtures from a real AMT: every read the driver makes, nothing that changes it."""
|
|
|
|
import pytest
|
|
import responses
|
|
|
|
from napalm_amt.record import READS, main, record
|
|
from napalm_amt.wsman import WsmanTransport
|
|
|
|
from .conftest import DMT, HOST, AmtServer
|
|
|
|
|
|
def test_records_every_read_under_the_fixture_names(amt, tmp_path):
|
|
record(WsmanTransport(HOST, "admin", "pw"), tmp_path)
|
|
names = sorted(p.name for p in tmp_path.iterdir())
|
|
assert "AMT_GeneralSettings.get.xml" in names
|
|
assert "CIM_PhysicalMemory.pull.xml" in names
|
|
assert "AMT_SetupAndConfigurationService.GetUuid.xml" in names
|
|
assert "AMT_MessageLog.GetRecords.xml" in names
|
|
assert "COMMANDS.txt" in names
|
|
assert (tmp_path / "CIM_PhysicalMemory.pull.xml").read_text() == (
|
|
DMT / "CIM_PhysicalMemory.pull.xml"
|
|
).read_text()
|
|
|
|
|
|
def test_a_second_page_gets_its_own_file(amt, tmp_path):
|
|
record(WsmanTransport(HOST, "admin", "pw"), tmp_path)
|
|
# CIM_Chip answers in two pages (the second is the mock's empty last page).
|
|
assert (tmp_path / "CIM_Chip.pull.xml").exists()
|
|
assert (tmp_path / "CIM_Chip.pull.2.xml").exists()
|
|
|
|
|
|
def test_never_switches_power_or_writes(amt, tmp_path):
|
|
record(WsmanTransport(HOST, "admin", "pw"), tmp_path)
|
|
operations = {r.operation for r in amt.requests}
|
|
assert operations <= {
|
|
"Get",
|
|
"Enumerate",
|
|
"Pull",
|
|
"GetUuid",
|
|
"PositionToFirstRecord",
|
|
"GetRecords",
|
|
}
|
|
|
|
|
|
def test_a_class_amt_lacks_is_recorded_as_its_fault(amt, tmp_path):
|
|
amt.override(
|
|
"CIM_MediaAccessDevice",
|
|
"Enumerate",
|
|
400,
|
|
(DMT / "fault_destination_unreachable.xml").read_text(),
|
|
)
|
|
record(WsmanTransport(HOST, "admin", "pw"), tmp_path)
|
|
assert (
|
|
"DestinationUnreachable" in (tmp_path / "CIM_MediaAccessDevice.enumerate.xml").read_text()
|
|
)
|
|
assert "CIM_MediaAccessDevice" in (tmp_path / "COMMANDS.txt").read_text()
|
|
|
|
|
|
def test_the_provisioning_otp_is_redacted(amt, tmp_path):
|
|
record(WsmanTransport(HOST, "admin", "pw"), tmp_path)
|
|
text = (tmp_path / "AMT_SetupAndConfigurationService.get.xml").read_text()
|
|
assert "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=" not in text
|
|
assert "<g:ProvisioningServerOTP>REDACTED</g:ProvisioningServerOTP>" in text
|
|
|
|
|
|
def test_every_read_names_a_known_class():
|
|
from napalm_amt.wsman import resource_uri
|
|
|
|
for resource, _operation in READS:
|
|
resource_uri(resource)
|
|
|
|
|
|
def test_main_reads_the_password_from_a_file(tmp_path, monkeypatch):
|
|
secret = tmp_path / "pw"
|
|
secret.write_text("P@ssw0rd\n")
|
|
out = tmp_path / "out"
|
|
with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
|
|
AmtServer(rsps)
|
|
assert (
|
|
main(
|
|
[
|
|
HOST,
|
|
"--user",
|
|
"admin",
|
|
"--password-file",
|
|
str(secret),
|
|
"--out",
|
|
str(out),
|
|
"--no-landing-page",
|
|
]
|
|
)
|
|
== 0
|
|
)
|
|
assert (out / "AMT_GeneralSettings.get.xml").exists()
|
|
|
|
|
|
def test_main_without_a_password_fails(tmp_path, monkeypatch):
|
|
monkeypatch.delenv("AMT_PASSWORD", raising=False)
|
|
with pytest.raises(SystemExit):
|
|
main([HOST, "--out", str(tmp_path)])
|
|
|
|
|
|
def test_landing_page_with_its_headers_is_kept(tmp_path):
|
|
with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
|
|
AmtServer(rsps)
|
|
rsps.add(
|
|
responses.GET,
|
|
f"http://{HOST}:16992/",
|
|
status=200,
|
|
headers={"Server": "Intel(R) Active Management Technology 11.8.50.3399"},
|
|
body="<html><title>Intel® Active Management Technology</title></html>",
|
|
)
|
|
secret = tmp_path / "pw"
|
|
secret.write_text("pw")
|
|
main([HOST, "--password-file", str(secret), "--out", str(tmp_path / "out")])
|
|
page = (tmp_path / "out" / "landing_page.txt").read_text()
|
|
assert "Server: Intel(R) Active Management Technology 11.8.50.3399" in page
|
|
assert "<title>Intel® Active Management Technology</title>" in page
|