[Core] Add get_safe_executable_path to prevent binary planting when spawning subprocess

doc/cli_subprocess_guidelines.md

@@ -68,9 +68,9 @@ All various kinds of `subprocess` Popen calling use cases can be easily adjusted
 
 #### Notes
 Besides that, developers might need to know some parts of the accessibility in both `run_cmd` and `subprocess`
-1. when calling shell built-in cmds, like `dir` or `echo`, using `shell=True` **in windows platform**, `subprocess` implicitly uses `cmd.exe`, while `run_cmd` asks developers to provide the `cmd.exe` as executable file specifically in the arg list's first item, like `["cmd.exe", "/c", "echo", "abc"]`
+1. when calling shell built-in cmds, like `dir` or `echo`, using `shell=True` **in windows platform**, `subprocess` implicitly uses `cmd.exe`, while `run_cmd` asks developers to provide the `cmd.exe` as executable file specifically in the arg list's first item, like `[get_safe_executable_path("cmd.exe"), "/c", "echo", "abc"]`
 2. if developers want to find an easy way to split their current cmd string into list, **for unix-like platforms**, developers can apply [`shlex.split`](https://docs.python.org/3/library/shlex.html#shlex.split) for quick access. But a prepared cmd statement is still more recommended (for more info about prepared cmd statement, please read below sections). 
-3. if developers want to locate the target command's executable file, a tool developers can use is `shutil.which` that gives the full executable file path in system, like `shutil.which(git)` returns the full `git.exe` path in windows platform `C:\\Program Files\\Git\\cmd\\git.EXE`. 
+3. to locate the target command's executable file, use `azure.cli.core.util.get_safe_executable_path` instead of `shutil.which`. It returns the full path (e.g. `C:\\Program Files\\Git\\cmd\\git.EXE` for `git`) or `None`, and never resolves from the current directory, which prevents binary planting (see [secure_executable_invocation.md](./secure_executable_invocation.md)).
 4. if the target cmd is `az`-related, like `az group show --name xxxx`, please see the following "`az`-related cmd execution" section.
 
 ### `az`-related command execution

doc/secure_executable_invocation.md

@@ -0,0 +1,49 @@
+# Secure Executable Invocation (Preventing Binary Planting)
+
+On Windows, a bare executable name (`cmd.exe`, `az`, `docker`, `git`, `gh`, `kubectl`, ...) is looked up in the
+**current working directory first**, both by `CreateProcess` (`shell=False`) and by `cmd.exe` (`shell=True`).
+If a user runs a command from an untrusted directory (e.g. a cloned repo) containing a planted binary, Azure CLI
+would execute it ([CWE-427](https://cwe.mitre.org/data/definitions/427.html)).
+
+```python
+# ❌ Vulnerable
+subprocess.check_output(['az', 'version'], shell=True)
+run_cmd(['docker', 'ps'])
+```
+
+## Fix
+
+Resolve the name to an absolute path with `azure.cli.core.util.get_safe_executable_path` before spawning it.
+It works like `shutil.which`, except that it:
+
+- searches `%SystemRoot%\System32` and `%SystemRoot%` first on Windows, then `PATH`;
+- skips the current working directory as well as relative `PATH` entries (`""`, `.`);
+- returns `None` when nothing trusted is found or the working directory is unknown. It never falls back to the
+  bare name.
+
+```python
+# ✅ Safe
+from azure.cli.core.azclierror import ValidationError
+from azure.cli.core.util import get_safe_executable_path, run_cmd
+
+docker = get_safe_executable_path('docker')
+if not docker:
+    raise ValidationError("Docker is not installed or not on PATH.")
+run_cmd([docker, 'ps'])
+```
+
+To run `az` itself (a batch script, `az.cmd`, on Windows), resolve both `cmd.exe` and `az`, and fall back to
+`[sys.executable, '-m', 'azure.cli']`. See `_get_az_cmd_prefix` in
+`src/azure-cli/azure/cli/command_modules/util/custom.py`. Prefer in-process calls over spawning `az`; see
+[cli_subprocess_guidelines.md](./cli_subprocess_guidelines.md).
+
+## Checklist for module owners
+
+1. Find bare-name spawns in your module:
+   ```bash
+   grep -rnE "(subprocess\.(run|call|check_call|check_output|Popen)|run_cmd)\(\s*\[?['\"](cmd\.exe|docker|git|gh|kubectl)" src/azure-cli/azure/cli/command_modules/<module>
+   ```
+2. Replace each bare name with `get_safe_executable_path('<name>')` and handle `None`.
+3. Pass arguments as a list and drop `shell=True`.
+4. Add a test that puts a decoy executable in the CWD and checks that it isn't selected (see
+   `TestSafeExecutablePath` in `src/azure-cli-core/azure/cli/core/tests/test_util.py`).

src/azure-cli-core/azure/cli/core/tests/test_util.py

@@ -17,7 +17,8 @@
     (get_file_json, truncate_text, shell_safe_json_parse, b64_to_hex, hash_string, random_string,
      open_page_in_browser, can_launch_browser, handle_exception, ConfiguredDefaultSetter, send_raw_request,
      should_disable_connection_verify, parse_proxy_resource_id, get_az_user_agent, get_az_rest_user_agent,
-    _get_parent_proc_name, is_wsl, run_cmd, run_az_cmd, roughly_parse_command, is_same_origin)
+    _get_parent_proc_name, is_wsl, run_cmd, run_az_cmd, roughly_parse_command, is_same_origin,
+     get_safe_executable_path)
 from azure.cli.core.mock import DummyCli
 
 
@@ -742,5 +743,55 @@ def test_roughly_parse_command(self):
         self.assertEqual(roughly_parse_command(['--help']), '')  # Starts with flag
         self.assertEqual(roughly_parse_command(['-h']), '')  # Starts with short flag
 
+class TestSafeExecutablePath(unittest.TestCase):
+    EXE = 'planted.exe' if platform.system() == 'Windows' else 'planted'
+
+    def setUp(self):
+        self._orig_cwd = os.getcwd()
+        self._tmp = tempfile.TemporaryDirectory()
+        self.cwd = self._tmp.name
+        os.chdir(self.cwd)
+        self._make_exe(self.cwd)  # decoy in the CWD
+
+    def tearDown(self):
+        os.chdir(self._orig_cwd)
+        self._tmp.cleanup()
+
+    def _make_exe(self, directory):
+        path = os.path.join(directory, self.EXE)
+        open(path, 'w').close()
+        os.chmod(path, 0o755)
+        return path
+
+    def _resolve(self, *path_dirs):
+        with mock.patch.dict(os.environ, {'PATH': os.pathsep.join(path_dirs)}):
+            return get_safe_executable_path(self.EXE)
+
+    def test_explicit_path_returned_as_is(self):
+        explicit = os.path.join('some', 'dir', 'tool')
+        self.assertEqual(get_safe_executable_path(explicit), explicit)
+
+    def test_cwd_never_resolved(self):
+        self.assertIsNone(self._resolve(self.cwd))
+        self.assertIsNone(self._resolve('.'))
+        if platform.system() == 'Windows':
+            self.assertIsNone(self._resolve(self.cwd.swapcase()))
+
+    def test_trusted_dir_used_after_cwd(self):
+        with tempfile.TemporaryDirectory() as path_dir:
+            target = self._make_exe(path_dir)
+            self.assertEqual(os.path.normcase(self._resolve(self.cwd, path_dir)), os.path.normcase(target))
+
+    def test_unknown_cwd_fails_closed(self):
+        with tempfile.TemporaryDirectory() as path_dir:
+            self._make_exe(path_dir)
+            with mock.patch('os.getcwd', side_effect=OSError):
+                self.assertIsNone(self._resolve(path_dir))
+
+    @unittest.skipUnless(platform.system() == 'Windows', 'Windows only')
+    def test_cmd_exe_resolves_to_system32(self):
+        resolved = get_safe_executable_path('cmd.exe')
+        self.assertTrue(resolved.lower().endswith(os.path.join('system32', 'cmd.exe')))
+
 if __name__ == '__main__':
     unittest.main()

src/azure-cli-core/azure/cli/core/util.py

@@ -1634,6 +1634,38 @@ def should_encrypt_token_cache(cli_ctx):
     return encrypt
 
 
+def get_safe_executable_path(executable):
+    """Resolve an executable name to an absolute path like ``shutil.which``, but never from the current working
+    directory, which Windows searches first for bare names (binary planting, CWE-427).
+    Return ``None`` if nothing trusted is found; never fall back to the bare name.
+    """
+    if os.path.dirname(executable):
+        return executable
+    try:
+        cwd = os.getcwd()
+    except OSError:
+        return None  # Fail closed: the directory to exclude is unknown.
+
+    search_dirs = os.environ.get('PATH', os.defpath).split(os.pathsep)
+    names = [executable]
+    if sys.platform.startswith('win'):
+        system_root = os.environ.get('SystemRoot', r'C:\Windows')
+        search_dirs = [os.path.join(system_root, 'System32'), system_root] + search_dirs
+        exts = [ext for ext in os.environ.get('PATHEXT', '.COM;.EXE;.BAT;.CMD').split(os.pathsep) if ext]
+        if os.path.splitext(executable)[1].upper() not in (ext.upper() for ext in exts):
+            names = [executable + ext for ext in exts]
+
+    for directory in (d.strip('"') for d in search_dirs):
+        # Relative entries ('', '.') resolve against the CWD as well.
+        if not os.path.isabs(directory) or not os.path.isdir(directory) or os.path.samefile(directory, cwd):
+            continue
+        for name in names:
+            path = os.path.join(directory, name)
+            if os.path.isfile(path) and os.access(path, os.X_OK):
+                return path
+    return None
+
+
 def run_cmd(args, *, capture_output=False, timeout=None, check=False, encoding=None, env=None):
     """Run command in a subprocess. It reduces (not eliminates) shell injection by forcing args to be a list
     and shell=False. Other arguments are keyword-only. For their documentation, see

src/azure-cli/azure/cli/command_modules/util/custom.py

@@ -181,7 +181,8 @@ def upgrade_version(cmd, update_all=None, yes=None, allow_preview=None):  # pyli
     importlib.reload(subprocess)
     importlib.reload(json)
 
-    version_result = subprocess.check_output(['az', 'version', '-o', 'json'], shell=platform.system() == 'Windows')
+    az_cmd_prefix = _get_az_cmd_prefix()
+    version_result = subprocess.check_output(az_cmd_prefix + ['version', '-o', 'json'])
     version_json = json.loads(version_result)
     new_version = version_json['azure-cli-core']
 
@@ -196,16 +197,34 @@ def upgrade_version(cmd, update_all=None, yes=None, allow_preview=None):  # pyli
     for ext_name in exts:
         try:
             logger.warning("Checking update for %s", ext_name)
-            cmds = ['az', 'extension', 'update', '-n', ext_name]
+            cmds = az_cmd_prefix + ['extension', 'update', '-n', ext_name]
             if allow_preview is not None:
                 cmds += ["--allow-preview", str(allow_preview)]
-            subprocess.call(cmds, shell=platform.system() == 'Windows')
+            subprocess.call(cmds)
         except Exception as ex:  # pylint: disable=broad-except
             msg = "Extension {} update failed during az upgrade. {}".format(ext_name, str(ex))
             raise CLIError(msg)
     logger.warning("Upgrade finished.")
 
 
+def _get_az_cmd_prefix():
+    """Return an absolute-path command prefix for spawning 'az', so nothing is resolved from the CWD (CWE-427).
+    On Windows, az.cmd runs through cmd.exe. Fall back to the current interpreter if either can't be resolved.
+    """
+    import platform
+    import sys
+    from azure.cli.core.util import get_safe_executable_path
+
+    az_path = get_safe_executable_path('az')
+    if platform.system() == 'Windows':
+        cmd_path = get_safe_executable_path('cmd.exe')
+        if az_path and cmd_path:
+            return [cmd_path, '/c', az_path]
+    elif az_path:
+        return [az_path]
+    return [sys.executable, '-m', 'azure.cli']
+
+
 def _upgrade_on_windows():
     """Download MSI to a temp folder and install it with msiexec.exe.
     Directly installing from URL may be blocked by policy: https://github.com/Azure/azure-cli/issues/19171
@@ -216,7 +235,7 @@ def _upgrade_on_windows():
     import sys
     import tempfile
 
-    from azure.cli.core.util import rmtree_with_retry
+    from azure.cli.core.util import rmtree_with_retry, get_safe_executable_path
 
     if platform.architecture()[0] == '32bit':
         msi_url = 'https://aka.ms/installazurecliwindows'
@@ -231,7 +250,10 @@ def _upgrade_on_windows():
 
     msi_path = _download_from_url(msi_url, msi_dir)
 
-    subprocess.Popen(['msiexec.exe', '/i', msi_path])
+    msiexec_path = get_safe_executable_path('msiexec.exe')
+    if not msiexec_path:
+        raise CLIError("Unable to locate msiexec.exe. Please install the MSI manually: {}".format(msi_path))
+    subprocess.Popen([msiexec_path, '/i', msi_path])
     logger.warning("Installation started. Please complete the upgrade in the opened window.\nTo update extensions, "
                    "please run `az upgrade` again after completing the upgrade.")
     sys.exit(0)