[3006.x] Remove wmic usage from disk grains on Windows

changelog/66959.fixed.md

@@ -0,0 +1,2 @@
+Removed the usage of wmic to get the disk and iscsi grains for Windows. The wmic
+binary is being deprecated.

salt/grains/disks.py

@@ -16,6 +16,7 @@
 __salt__ = {
     "cmd.run": salt.modules.cmdmod._run_quiet,
     "cmd.run_all": salt.modules.cmdmod._run_all_quiet,
+    "cmd.powershell": salt.modules.cmdmod.powershell,
 }
 
 log = logging.getLogger(__name__)
@@ -153,41 +154,28 @@ def _linux_disks():
 
 
 def _windows_disks():
-    wmic = salt.utils.path.which("wmic")
-
-    namespace = r"\\root\microsoft\windows\storage"
-    path = "MSFT_PhysicalDisk"
-    get = "DeviceID,MediaType"
 
+    cmd = "Get-PhysicalDisk | Select DeviceID, MediaType"
     ret = {"disks": [], "ssds": []}
 
-    cmdret = __salt__["cmd.run_all"](
-        "{} /namespace:{} path {} get {} /format:table".format(
-            wmic, namespace, path, get
-        )
-    )
-
-    if cmdret["retcode"] != 0:
-        log.trace("Disk grain does not support this version of Windows")
-    else:
-        for line in cmdret["stdout"].splitlines():
-            info = line.split()
-            if len(info) != 2 or not info[0].isdigit() or not info[1].isdigit():
-                continue
-            device = rf"\\.\PhysicalDrive{info[0]}"
-            mediatype = info[1]
-            if mediatype == "3":
-                log.trace("Device %s reports itself as an HDD", device)
-                ret["disks"].append(device)
-            elif mediatype == "4":
-                log.trace("Device %s reports itself as an SSD", device)
-                ret["ssds"].append(device)
-                ret["disks"].append(device)
-            elif mediatype == "5":
-                log.trace("Device %s reports itself as an SCM", device)
-                ret["disks"].append(device)
-            else:
-                log.trace("Device %s reports itself as Unspecified", device)
-                ret["disks"].append(device)
+    drive_info = __salt__["cmd.powershell"](cmd)
+
+    if not drive_info:
+        log.trace("No physical discs found")
+        return ret
+
+    # We need a list of dict
+    if isinstance(drive_info, dict):
+        drive_info = [drive_info]
+
+    for drive in drive_info:
+        # Make sure we have a valid drive type
+        if drive["MediaType"].lower() not in ["hdd", "ssd", "scm", "unspecified"]:
+            log.trace(f'Unknown media type: {drive["MediaType"]}')
+            continue
+        device = rf'\\.\PhysicalDrive{drive["DeviceID"]}'
+        ret["disks"].append(device)
+        if drive["MediaType"].lower() == "ssd":
+            ret["ssds"].append(device)
 
     return ret

salt/grains/iscsi.py

@@ -85,28 +85,25 @@ def _aix_iqn():
 
 def _windows_iqn():
     """
-    Return iSCSI IQN from a Windows host.
+    Return iSCSI nodes from a Windows host.
     """
+    cmd = "Get-InitiatorPort | Select NodeAddress"
     ret = []
 
-    wmic = salt.utils.path.which("wmic")
+    nodes = salt.modules.cmdmod.powershell(cmd)
 
-    if not wmic:
+    if not nodes:
+        log.trace("No iSCSI nodes found")
         return ret
 
-    namespace = r"\\root\WMI"
-    path = "MSiSCSIInitiator_MethodClass"
-    get = "iSCSINodeName"
+    # A single node will return a dictionary with a single entry
+    # {"NodeAddress": "iqn.1991-05.com.microsoft:johnj99-pc2.contoso.com"}
+    # Multiple nodes will return a list of single entry dicts
+    # We need a list of dict
+    if isinstance(nodes, dict):
+        nodes = [nodes]
 
-    cmd_ret = salt.modules.cmdmod.run_all(
-        "{} /namespace:{} path {} get {} /format:table".format(
-            wmic, namespace, path, get
-        )
-    )
-
-    for line in cmd_ret["stdout"].splitlines():
-        if line.startswith("iqn."):
-            line = line.rstrip()
-            ret.append(line.rstrip())
+    for node in nodes:
+        ret.append(node["NodeAddress"])
 
     return ret

salt/modules/cmdmod.py

@@ -266,7 +266,7 @@ def _prep_powershell_cmd(win_shell, cmd, encoded_cmd):
     win_shell = salt.utils.path.which(win_shell)
 
     if not win_shell:
-        raise CommandExecutionError("PowerShell binary not found")
+        raise CommandExecutionError(f"PowerShell binary not found: {win_shell}")
 
     new_cmd = [win_shell, "-NonInteractive", "-NoProfile", "-ExecutionPolicy", "Bypass"]
 

salt/utils/path.py

@@ -203,7 +203,7 @@ def is_executable(path, membership=res):
 
     # now to search through our system_path
     for path in system_path:
-        p = join(path, exe)
+        p = join(os.path.expandvars(path), exe)
 
         # iterate through all extensions to see which one is executable
         for ext in pathext:

tests/pytests/unit/grains/test_disks.py

@@ -2,8 +2,6 @@
     :codeauthor: :email:`Shane Lee <[email protected]>`
 """
 
-import textwrap
-
 import pytest
 
 import salt.grains.disks as disks
@@ -17,63 +15,58 @@ def configure_loader_modules():
     }
 
 
-def test__windows_disks():
+def test__windows_disks_dict():
     """
-    Test grains._windows_disks, normal return
-    Should return a populated dictionary
+    Test grains._windows_disks with a single disk returned as a dict
+    Should return 1 disk and no ssds
+    """
+    devices = {"DeviceID": 0, "MediaType": "HDD"}
+    mock_powershell = MagicMock(return_value=devices)
+
+    with patch.dict(disks.__salt__, {"cmd.powershell": mock_powershell}):
+        result = disks._windows_disks()
+        expected = {"disks": ["\\\\.\\PhysicalDrive0"], "ssds": []}
+        assert result == expected
+
+
+def test__windows_disks_list():
     """
-    mock_which = MagicMock(return_value="C:\\Windows\\System32\\wbem\\WMIC.exe")
-    wmic_result = textwrap.dedent(
-        """
-        DeviceId  MediaType
-        0         4
-        1         0
-        2         3
-        3         5
+    test grains._windows_disks with multiple disks and types as a list of dicts
+    Should return 4 disks and 1 ssd
     """
-    )
-    mock_run_all = MagicMock(return_value={"stdout": wmic_result, "retcode": 0})
+    devices = [
+        {"DeviceID": 0, "MediaType": "SSD"},
+        {"DeviceID": 1, "MediaType": "HDD"},
+        {"DeviceID": 2, "MediaType": "HDD"},
+        {"DeviceID": 3, "MediaType": "HDD"},
+    ]
+    mock_powershell = MagicMock(return_value=devices)
 
-    with patch("salt.utils.path.which", mock_which), patch.dict(
-        disks.__salt__, {"cmd.run_all": mock_run_all}
-    ):
+    with patch.dict(disks.__salt__, {"cmd.powershell": mock_powershell}):
         result = disks._windows_disks()
         expected = {
-            "ssds": ["\\\\.\\PhysicalDrive0"],
             "disks": [
                 "\\\\.\\PhysicalDrive0",
                 "\\\\.\\PhysicalDrive1",
                 "\\\\.\\PhysicalDrive2",
                 "\\\\.\\PhysicalDrive3",
             ],
+            "ssds": ["\\\\.\\PhysicalDrive0"],
         }
         assert result == expected
-        cmd = " ".join(
-            [
-                "C:\\Windows\\System32\\wbem\\WMIC.exe",
-                "/namespace:\\\\root\\microsoft\\windows\\storage",
-                "path",
-                "MSFT_PhysicalDisk",
-                "get",
-                "DeviceID,MediaType",
-                "/format:table",
-            ]
-        )
-        mock_run_all.assert_called_once_with(cmd)
-
-
-def test__windows_disks_retcode():
+
+
+def test__windows_disks_empty():
     """
-    Test grains._windows_disks, retcode 1
+    Test grains._windows_disks when nothing is returned
     Should return empty lists
     """
-    mock_which = MagicMock(return_value="C:\\Windows\\System32\\wbem\\WMIC.exe")
-    mock_run_all = MagicMock(return_value={"stdout": "", "retcode": 1})
-    with patch("salt.utils.path.which", mock_which), patch.dict(
-        disks.__salt__, {"cmd.run_all": mock_run_all}
-    ):
+    devices = {}
+    mock_powershell = MagicMock(return_value=devices)
+
+    with patch.dict(disks.__salt__, {"cmd.powershell": mock_powershell}):
+        expected = {"disks": [], "ssds": []}
         result = disks._windows_disks()
-        expected = {"ssds": [], "disks": []}
         assert result == expected
 
 

tests/pytests/unit/grains/test_iscsi.py

@@ -9,16 +9,39 @@
 from tests.support.mock import MagicMock, mock_open, patch
 
 
-def test_windows_iscsi_iqn_grains():
-    cmd_run_mock = MagicMock(
-        return_value={"stdout": "iSCSINodeName\niqn.1991-05.com.microsoft:simon-x1\n"}
-    )
-    _grains = {}
-    with patch("salt.utils.path.which", MagicMock(return_value=True)):
-        with patch("salt.modules.cmdmod.run_all", cmd_run_mock):
-            _grains["iscsi_iqn"] = iscsi._windows_iqn()
-
-    assert _grains.get("iscsi_iqn") == ["iqn.1991-05.com.microsoft:simon-x1"]
+def test_windows_iscsi_iqn_grains_empty():
+    nodes_dict = {}
+    cmd_powershell_mock = MagicMock(return_value=nodes_dict)
+    with patch("salt.modules.cmdmod.powershell", cmd_powershell_mock):
+        result = iscsi._windows_iqn()
+    expected = []
+    assert result == expected
+
+
+def test_windows_iscsi_iqn_grains_single():
+    nodes_dict = {"NodeAddress": "iqn.1991-05.com.microsoft:simon-x1"}
+    cmd_powershell_mock = MagicMock(return_value=nodes_dict)
+    with patch("salt.modules.cmdmod.powershell", cmd_powershell_mock):
+        result = iscsi._windows_iqn()
+    expected = ["iqn.1991-05.com.microsoft:simon-x1"]
+    assert result == expected
+
+
+def test_windows_iscsi_iqn_grains_multiple():
+    nodes_list = [
+        {"NodeAddress": "iqn.1991-05.com.microsoft:simon-x1"},
+        {"NodeAddress": "iqn.1991-05.com.microsoft:simon-x2"},
+        {"NodeAddress": "iqn.1991-05.com.microsoft:simon-x3"},
+    ]
+    cmd_powershell_mock = MagicMock(return_value=nodes_list)
+    with patch("salt.modules.cmdmod.powershell", cmd_powershell_mock):
+        result = iscsi._windows_iqn()
+    expected = [
+        "iqn.1991-05.com.microsoft:simon-x1",
+        "iqn.1991-05.com.microsoft:simon-x2",
+        "iqn.1991-05.com.microsoft:simon-x3",
+    ]
+    assert result == expected
 
 
 def test_aix_iscsi_iqn_grains():