{Core} Fix injection risk during extension dynamic installation (#33990)

Author: necusjz

src/azure-cli-core/azure/cli/core/extension/dynamic_install.py

@@ -241,13 +241,17 @@ def _check_value_in_extensions(cli_ctx, parser, args, no_prompt):  # pylint: dis
         from azure.cli.core.extension.operations import add_extension
         add_extension(cli_ctx=cli_ctx, extension_name=ext_name, upgrade=True, allow_preview=extension_allow_preview)
         if run_after_extension_installed:
-            import subprocess
-            import platform
-            exit_code = subprocess.call(args, shell=platform.system() == 'Windows')
+            # Rerun the command with the CLI's own entry point instead of handing the raw argv back to a
+            # shell. The previous implementation used `subprocess.call(args, shell=True)` on Windows, which
+            # let cmd.exe reinterpret user-controlled arguments (e.g. `&`, `|`, `^`) and allowed command
+            # injection. `run_az_cmd` executes the command in-process with the args kept as a list, so no
+            # shell is involved.
+            from azure.cli.core.util import run_az_cmd
+            result = run_az_cmd(args[1:])
+            exit_code = getattr(result, 'exit_code', 0)
             # In this case, error msg is for telemetry recording purpose only.
-            # From UX perspective, the command will rerun in subprocess. Whether it succeeds or fails,
-            # mesages will be shown from the subprocess and this process should not print more message to
-            # interrupt that.
+            # From UX perspective, the command is rerun and its messages are already shown,
+            # so this process should not print more message to interrupt that.
             print_error = False
             error_msg = ("Extension {} dynamically installed{} and commands will be "
                          "rerun automatically.").format(ext_name, prompt_info)

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

@@ -274,6 +274,34 @@ def mock_add_extension(*args, **kwargs):
                 parser.parse_args('test new-ext reset pos1 pos2'.split())  # test positional args
             self.assertIn("Extension another-ext-name installed. Please rerun your command.", logger_msgs[6])
 
+        # test dynamic extension install with the command rerun after installation.
+        # The command must be rerun via run_az_cmd (args as a list, no shell) to avoid command injection.
+        from knack.util import CommandResultItem
+        run_az_cmd_args = []
+
+        def mock_run_az_cmd(args, out_file=None):  # pylint: disable=unused-argument
+            run_az_cmd_args.append(args)
+            return CommandResultItem(None, exit_code=0)
+
+        with mock.patch.object(logging.Logger, 'error', mock_log_error), \
+                mock.patch.object(azure.cli.core.extension.operations, 'add_extension', mock_add_extension), \
+                mock.patch.object(azure.cli.core.extension.dynamic_install, '_get_extension_command_tree',
+                                  mock_ext_cmd_tree_load), \
+                mock.patch.object(azure.cli.core.extension.dynamic_install, '_get_extension_use_dynamic_install_config',
+                                  return_value='yes_without_prompt'), \
+                mock.patch.object(azure.cli.core.extension.dynamic_install,
+                                  '_get_extension_run_after_dynamic_install_config', return_value=True), \
+                mock.patch.object(azure.cli.core.util, 'run_az_cmd', mock_run_az_cmd):
+            with self.assertRaises(SystemExit) as cm:
+                parser.parse_args(['test', 'new-ext', 'create', '--opt', 'enum_2 & echo injected'])
+            # exit code comes from the rerun result
+            self.assertEqual(cm.exception.code, 0)
+            # args are passed as a list without the 'az' entry point, so no shell can reinterpret them
+            self.assertEqual(run_az_cmd_args,
+                             [['test', 'new-ext', 'create', '--opt', 'enum_2 & echo injected']])
+            # error message is only for telemetry, it should not be printed again
+            self.assertEqual(len(logger_msgs), 7)
+
 
 class VerifyError:  # pylint: disable=too-few-public-methods