Add Redfish transport

Session login through the SessionService with X-Auth-Token, the session deleted on close (also on errors and when the DELETE fails), basic auth as fallback only when the service cannot create sessions, never after a rejected password. Certificate verification is on by default and configurable per device; True passes the requests CA bundle explicitly because the redfish library disables verification without a CA file.

Collections are followed by @odata.id, including expanded members and nextLink pages. Links to other hosts are refused so the session token stays on the BMC.

tests/test_transport.py

@@ -0,0 +1,428 @@
+"""Transport tests against mocked HTTP serving the DMTF public-rackmount1 mockup."""
+
+import base64
+import json
+import logging
+import os
+
+import pytest
+import requests
+from conftest import BASE_URL, SESSION_PATH, SESSIONS, TOKEN
+
+from napalm_redfish.transport import (
+    RedfishAuthError,
+    RedfishError,
+    RedfishHTTPError,
+    RedfishNotFound,
+    RedfishTransport,
+)
+
+BASIC = "Basic " + base64.b64encode(b"netork:s3cret").decode()
+
+
+def make(**kwargs) -> RedfishTransport:
+    kwargs.setdefault("verify", False)
+    kwargs.setdefault("retries", 0)
+    return RedfishTransport("bmc.example.net", "netork", "s3cret", **kwargs)
+
+
+# --- base URL ------------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "hostname, port, expected",
+    [
+        ("bmc.example.net", None, "https://bmc.example.net"),
+        ("bmc.example.net", 8443, "https://bmc.example.net:8443"),
+        ("192.0.2.10", None, "https://192.0.2.10"),
+        ("2001:db8::10", None, "https://[2001:db8::10]"),
+        ("2001:db8::10", 8443, "https://[2001:db8::10]:8443"),
+    ],
+)
+def test_base_url(hostname, port, expected):
+    assert RedfishTransport(hostname, "u", "p", port=port).base_url == expected
+
+
+def test_rejects_unknown_auth_method():
+    with pytest.raises(ValueError):
+        RedfishTransport("bmc.example.net", "u", "p", auth="digest")
+
+
+# --- session login -------------------------------------------------------------------------
+
+
+def test_open_reads_service_root_before_login(server):
+    t = make()
+    t.open()
+    first = server.rsps.calls[0].request
+    assert first.method == "GET" and first.url.rstrip("/") == BASE_URL + "/redfish/v1"
+    assert "X-Auth-Token" not in first.headers and "Authorization" not in first.headers
+    assert t.root["RedfishVersion"] == "1.15.0"
+    t.close()
+
+
+def test_session_login_posts_credentials_and_uses_token(server):
+    t = make()
+    t.open()
+    assert t.auth_method == "session"
+    (login,) = server.calls("POST", SESSIONS)
+    assert json.loads(login.request.body) == {"UserName": "netork", "Password": "s3cret"}
+
+    t.get("/redfish/v1/Systems")
+    get = server.calls("GET", "/redfish/v1/Systems")[0].request
+    assert get.headers["X-Auth-Token"] == TOKEN
+    assert "Authorization" not in get.headers
+    t.close()
+
+
+def test_close_deletes_the_session(server):
+    t = make()
+    t.open()
+    t.close()
+    (delete,) = server.calls("DELETE", SESSION_PATH)
+    assert delete.request.headers["X-Auth-Token"] == TOKEN
+    assert t.auth_method is None
+
+
+def test_open_twice_logs_in_once(server):
+    t = make()
+    t.open()
+    t.open()
+    t.close()
+    assert len(server.calls("POST", SESSIONS)) == 1
+    assert len(server.calls("DELETE", SESSION_PATH)) == 1
+
+
+def test_close_twice_deletes_once(server):
+    t = make()
+    t.open()
+    t.close()
+    t.close()
+    assert len(server.calls("DELETE")) == 1
+
+
+def test_context_manager_deletes_session_on_error(server):
+    with pytest.raises(RuntimeError):
+        with make() as t:
+            t.get("/redfish/v1/Systems")
+            raise RuntimeError("boom")
+    assert len(server.calls("DELETE", SESSION_PATH)) == 1
+
+
+def test_close_survives_failed_logout(server, caplog):
+    server.override("DELETE", SESSION_PATH, 500, {"error": {"message": "nope"}})
+    t = make()
+    t.open()
+    with caplog.at_level(logging.WARNING):
+        t.close()
+    assert t.auth_method is None
+    assert "session" in caplog.text.lower()
+
+
+@pytest.mark.filterwarnings("ignore:Service incorrectly did not provide the 'Location'")
+def test_session_uri_from_body_when_location_missing(server):
+    server.override("POST", SESSIONS, 201, {"@odata.id": SESSION_PATH}, {"X-Auth-Token": TOKEN})
+    t = make()
+    t.open()
+    t.close()
+    assert len(server.calls("DELETE", SESSION_PATH)) == 1
+
+
+def test_wrong_password_raises_without_basic_retry(server):
+    server.override("POST", SESSIONS, 401, {"error": {"message": "bad credentials"}})
+    t = make()
+    with pytest.raises(RedfishAuthError):
+        t.open()
+    # One failed login only: a second attempt with basic auth would count twice against
+    # the BMC's lockout threshold.
+    assert len(server.calls("POST")) == 1
+    assert not [c for c in server.rsps.calls if "Authorization" in c.request.headers]
+
+
+@pytest.mark.parametrize("status", [404, 405, 501])
+def test_falls_back_to_basic_when_sessions_unsupported(server, status):
+    server.override("POST", SESSIONS, status, {"error": {"message": "not supported"}})
+    t = make()
+    t.open()
+    assert t.auth_method == "basic"
+    t.get("/redfish/v1/Systems")
+    get = server.calls("GET", "/redfish/v1/Systems")[0].request
+    assert get.headers["Authorization"] == BASIC
+    assert "X-Auth-Token" not in get.headers
+    t.close()
+    assert not server.calls("DELETE")
+
+
+@pytest.mark.filterwarnings("ignore:Service incorrectly did not provide the 'Location'")
+def test_falls_back_to_basic_when_no_token_returned(server):
+    server.override("POST", SESSIONS, 201, {"@odata.id": SESSION_PATH})
+    t = make()
+    t.open()
+    assert t.auth_method == "basic"
+    t.close()
+
+
+def test_no_fallback_when_session_auth_is_forced(server):
+    server.override("POST", SESSIONS, 405, {})
+    t = make(auth="session")
+    with pytest.raises(RedfishAuthError):
+        t.open()
+
+
+def test_basic_auth_never_creates_a_session(server):
+    t = make(auth="basic")
+    t.open()
+    assert t.auth_method == "basic"
+    t.get("/redfish/v1/Systems")
+    assert not server.calls("POST")
+    assert server.calls("GET", "/redfish/v1/Systems")[0].request.headers["Authorization"] == BASIC
+    t.close()
+
+
+def test_basic_auth_wrong_password(server):
+    server.override("GET", SESSIONS, 401, {})
+    with pytest.raises(RedfishAuthError):
+        make(auth="basic").open()
+
+
+def test_login_url_from_service_root(server):
+    custom = dict(_root(), Links={"Sessions": {"@odata.id": "/redfish/v1/Custom/Sessions"}})
+    server.override("GET", "/redfish/v1/", 200, custom)
+    server.override("GET", "/redfish/v1", 200, custom)
+    server.override("POST", "/redfish/v1/Custom/Sessions", 201, {},
+                    {"X-Auth-Token": TOKEN, "Location": SESSION_PATH})
+    t = make()
+    t.open()
+    assert server.calls("POST", "/redfish/v1/Custom/Sessions")
+    t.close()
+
+
+def _root():
+    from conftest import load
+
+    return load("dmtf-public-rackmount1", "/redfish/v1")
+
+
+def test_unreachable_service_raises_redfish_error(http):
+    t = make()
+    with pytest.raises(RedfishError):
+        t.open()  # nothing registered on the mock: connection refused
+
+
+def test_requests_before_open_fail(server):
+    with pytest.raises(RedfishError):
+        make().get("/redfish/v1/Systems")
+
+
+# --- certificate verification --------------------------------------------------------------
+
+
+def _verify_of(server):
+    return server.rsps.calls[0].request.req_kwargs["verify"]
+
+
+def test_verify_true_uses_ca_bundle(server):
+    t = make(verify=True)
+    t.open()
+    verify = _verify_of(server)
+    # The redfish library turns verification off unless it gets a CA file, so True must
+    # become a real bundle, not None/False.
+    assert isinstance(verify, str) and os.path.isfile(verify)
+    assert verify == requests.certs.where()
+    t.close()
+
+
+def test_verify_false_disables_verification(server):
+    t = make(verify=False)
+    t.open()
+    assert _verify_of(server) is False
+    t.close()
+
+
+def test_verify_path_is_passed_through(server, tmp_path):
+    ca = tmp_path / "bmc-ca.pem"
+    ca.write_text("dummy")
+    t = make(verify=str(ca))
+    t.open()
+    assert _verify_of(server) == str(ca)
+    t.close()
+
+
+def test_verify_path_must_exist():
+    with pytest.raises(ValueError):
+        RedfishTransport("bmc.example.net", "u", "p", verify="/nonexistent/ca.pem")
+
+
+def test_verify_is_the_default():
+    assert RedfishTransport("bmc.example.net", "u", "p").verify is True
+
+
+# --- GET and errors ------------------------------------------------------------------------
+
+
+@pytest.fixture
+def transport(server):
+    t = make()
+    t.open()
+    yield t
+    t.close()
+
+
+def test_get_returns_json(transport):
+    assert transport.get("/redfish/v1/Systems/437XR1138R2")["Model"] == "3500"
+
+
+def test_get_accepts_odata_reference(transport):
+    ref = {"@odata.id": "/redfish/v1/Systems/437XR1138R2"}
+    assert transport.get(ref)["Id"] == "437XR1138R2"
+
+
+def test_get_strips_fragment(transport, server):
+    data = transport.get("/redfish/v1/Chassis/1U/Thermal#/Fans/0")
+    assert data["Id"] == "Thermal"
+    assert server.calls("GET", "/redfish/v1/Chassis/1U/Thermal")[0].request.url.endswith("/Thermal")
+
+
+def test_get_accepts_same_host_absolute_url(transport):
+    assert transport.get(BASE_URL + "/redfish/v1/Systems")["Members@odata.count"] == 1
+
+
+def test_get_refuses_links_to_other_hosts(transport, server):
+    before = len(server.rsps.calls)
+    # PowerSupply.Links.Outlet in the mockup points at a PDU on another host. Following it
+    # would send this BMC's session token there.
+    with pytest.raises(RedfishError):
+        transport.get("https://redfishpdu.contoso.com/redfish/v1/PowerEquipment/RackPDUs/1")
+    assert len(server.rsps.calls) == before
+
+
+def test_get_refuses_relative_paths(transport):
+    with pytest.raises(RedfishError):
+        transport.get("Systems/1")
+
+
+def test_get_404_raises_not_found(transport):
+    with pytest.raises(RedfishNotFound) as exc:
+        transport.get("/redfish/v1/Systems/1")
+    assert exc.value.status == 404
+    assert exc.value.path == "/redfish/v1/Systems/1"
+
+
+def test_get_optional_returns_none_on_404(transport):
+    assert transport.get_optional("/redfish/v1/Chassis/1U/Missing") is None
+    assert transport.get_optional(None) is None
+
+
+def test_get_error_carries_redfish_message(transport, server):
+    server.override("GET", "/redfish/v1/Managers", 500, {
+        "error": {"code": "Base.1.0.InternalError", "message": "General error",
+                  "@Message.ExtendedInfo": [{"Message": "The BMC is busy."}]}})
+    with pytest.raises(RedfishHTTPError) as exc:
+        transport.get("/redfish/v1/Managers")
+    assert exc.value.status == 500
+    assert "The BMC is busy." in str(exc.value)
+
+
+@pytest.mark.parametrize("status", [401, 403])
+def test_get_auth_errors(transport, server, status):
+    server.override("GET", "/redfish/v1/AccountService", status, {})
+    with pytest.raises(RedfishAuthError):
+        transport.get("/redfish/v1/AccountService")
+
+
+def test_get_invalid_json(transport, server):
+    server.override("GET", "/redfish/v1/Managers", 200, "<html>login</html>")
+    with pytest.raises(RedfishError):
+        transport.get("/redfish/v1/Managers")
+
+
+# --- collections ---------------------------------------------------------------------------
+
+
+def test_members_follows_odata_ids(transport, server):
+    systems = transport.members("/redfish/v1/Systems")
+    # The mockup's system ID is its serial number, not "1".
+    assert [s["Id"] for s in systems] == ["437XR1138R2"]
+    assert server.calls("GET", "/redfish/v1/Systems/437XR1138R2")
+
+
+def test_members_ignores_member_count(transport):
+    # Members@odata.count says 2, but three members are listed: trust the list.
+    fw = transport.members("/redfish/v1/UpdateService/FirmwareInventory")
+    assert [f["Id"] for f in fw] == ["BMC", "SS", "BIOS"]
+
+
+def test_members_from_link_on_resource(transport):
+    system = transport.get("/redfish/v1/Systems/437XR1138R2")
+    cpus = transport.members(system["Processors"])
+    assert [c["Id"] for c in cpus] == ["CPU1", "CPU2", "FPGA1"]
+
+
+def test_members_uses_expanded_members_without_extra_requests(transport, server):
+    entries = transport.members("/redfish/v1/Managers/BMC/LogServices/Log/Entries")
+    assert entries[0]["Message"] == "System May be Melting"
+    assert not server.calls("GET", "/redfish/v1/Managers/BMC/LogServices/Log/Entries/1")
+
+
+def test_members_follows_next_link(transport, server):
+    path = "/redfish/v1/Systems/437XR1138R2/LogServices/Log1/Entries"
+    server.override("GET", path + "?$skiptoken=2", 200, {
+        "Members": [{"@odata.id": path + "/3", "Id": "3", "Message": "third"}],
+        "@odata.id": path,
+    })
+    entries = transport.members(path)
+    assert [e["Id"] for e in entries] == ["1", "2", "3"]
+
+
+def test_members_next_link_loop_terminates(transport, server):
+    path = "/redfish/v1/Managers"
+    server.override("GET", path, 200, {
+        "Members": [{"@odata.id": "/redfish/v1/Managers/BMC"}],
+        "Members@odata.nextLink": path,
+    })
+    assert [m["Id"] for m in transport.members(path)] == ["BMC"]
+
+
+def test_members_limit(transport, server):
+    path = "/redfish/v1/Systems/437XR1138R2/LogServices/Log1/Entries"
+    entries = transport.members(path, limit=1)
+    assert [e["Id"] for e in entries] == ["1"]
+    assert len(server.calls("GET", path)) == 1
+
+
+def test_members_of_missing_collection(transport):
+    with pytest.raises(RedfishNotFound):
+        transport.members("/redfish/v1/Systems/437XR1138R2/Storage")
+
+
+def test_follow(transport):
+    manager = transport.get("/redfish/v1/Managers/BMC")
+    assert transport.follow(manager, "NetworkProtocol")["Id"] == "NetworkProtocol"
+    assert transport.follow(manager, "NoSuchLink") is None
+
+
+def test_follow_missing_target_is_none(transport, server):
+    system = transport.get("/redfish/v1/Systems/437XR1138R2")
+    # Storage is not linked from this system; Bios is linked but not in the fixtures.
+    assert transport.follow(system, "Storage") is None
+    assert transport.follow(system, "Bios") is None
+
+
+# --- POST ----------------------------------------------------------------------------------
+
+
+def test_post_sends_json_and_returns_body(transport, server):
+    target = "/redfish/v1/Systems/437XR1138R2/Actions/ComputerSystem.Reset"
+    server.override("POST", target, 204, {})
+    assert transport.post(target, {"ResetType": "GracefulRestart"}) == {}
+    (call,) = server.calls("POST", target)
+    assert json.loads(call.request.body) == {"ResetType": "GracefulRestart"}
+    assert call.request.headers["X-Auth-Token"] == TOKEN
+
+
+def test_post_error(transport, server):
+    target = "/redfish/v1/Managers/BMC/Actions/Manager.Reset"
+    server.override("POST", target, 400, {"error": {"message": "ActionParameterValueNotInList"}})
+    with pytest.raises(RedfishHTTPError) as exc:
+        transport.post(target, {"ResetType": "PowerCycle"})
+    assert exc.value.status == 400