{Storage} Secure AzCopy executable resolution

src/azure-cli/azure/cli/command_modules/storage/azcopy/util.py

@@ -5,6 +5,8 @@
 
 import os
 import platform
+import re
+import shutil
 import subprocess
 import datetime
 import stat
@@ -14,7 +16,7 @@
 from azure.cli.core.api import get_config_dir
 from knack.log import get_logger
 from knack.util import CLIError
-from packaging.version import parse as parse_version
+from packaging.version import InvalidVersion, parse as parse_version
 
 logger = get_logger(__name__)
 
@@ -27,24 +29,20 @@
 class AzCopy:
     def __init__(self, creds=None):
         self.system = platform.system()
-        install_location = _get_default_install_location()
+        install_location = os.path.abspath(_get_default_install_location())
         self.executable = None
         if os.path.isfile(install_location):
             self.executable = install_location
             version = self.check_version()
-            if not version or parse_version(version) < parse_version(AZCOPY_VERSION):
+            if not _is_supported_version(version):
                 self.executable = None
         else:
-            try:
-                import re
-                args = ["azcopy", "--version"]
-                out_bytes = subprocess.check_output(args)
-                out_text = out_bytes.decode('utf-8')
-                version = re.findall(r"azcopy version (.+?)\n", out_text)[0]
-                if version and parse_version(version) >= parse_version(AZCOPY_VERSION):
-                    self.executable = "azcopy"
-            except Exception:  # pylint: disable=broad-except
-                self.executable = None
+            system_executable = _resolve_system_executable("azcopy")
+            if system_executable:
+                self.executable = system_executable
+                version = self.check_version()
+                if not _is_supported_version(version):
+                    self.executable = None
         self.creds = creds
         if not self.executable:
             logger.warning("Azcopy not found, installing at %s", install_location)
@@ -93,13 +91,12 @@ def install_azcopy(self, install_location):
 
     def check_version(self):
         try:
-            import re
             args = [self.executable] + ["--version"]
             out_bytes = subprocess.check_output(args)
             out_text = out_bytes.decode('utf-8')
-            version = re.findall(r"azcopy version (.+?)\n", out_text)[0]
-            return version
-        except subprocess.CalledProcessError:
+            match = re.search(r"azcopy version (.+?)(?:\r?\n|$)", out_text)
+            return match.group(1) if match else ""
+        except (OSError, subprocess.CalledProcessError, UnicodeError):
             return ""
 
     def run_command(self, args):
@@ -130,6 +127,44 @@ def sync(self, source, destination, flags=None):
         self.run_command(['sync', source, destination] + flags)
 
 
+def _resolve_system_executable(executable):
+    resolved_path = shutil.which(executable)
+    if not resolved_path:
+        return None
+
+    try:
+        cwd = os.path.abspath(os.getcwd())
+        resolved_path = os.path.abspath(resolved_path)
+        resolved_dirs = (
+            os.path.dirname(resolved_path),
+            os.path.dirname(os.path.realpath(resolved_path)),
+        )
+    except (OSError, ValueError):
+        logger.warning("Ignoring %s because its path could not be validated.", executable)
+        return None
+
+    for resolved_dir in resolved_dirs:
+        try:
+            is_cwd = os.path.samefile(resolved_dir, cwd)
+        except (OSError, ValueError):
+            is_cwd = (os.path.normcase(os.path.normpath(resolved_dir)) ==
+                      os.path.normcase(os.path.normpath(cwd)))
+        if is_cwd:
+            logger.warning("Ignoring %s found in the current working directory.", executable)
+            return None
+
+    return resolved_path
+
+
+def _is_supported_version(version):
+    if not version:
+        return False
+    try:
+        return parse_version(version) >= parse_version(AZCOPY_VERSION)
+    except InvalidVersion:
+        return False
+
+
 class AzCopyCredentials:  # pylint: disable=too-few-public-methods
     def __init__(self, sas_token=None, token_info=None, tenant_id=None):
         self.sas_token = sas_token

src/azure-cli/azure/cli/command_modules/storage/tests/latest/test_storage_azcopy_util.py

@@ -0,0 +1,103 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import os
+import tempfile
+import unittest
+from unittest import mock
+
+from azure.cli.command_modules.storage.azcopy import util
+
+
+class TestAzCopyExecutableResolution(unittest.TestCase):
+
+    def test_cwd_executable_is_not_probed_or_reused(self):
+        with tempfile.TemporaryDirectory() as cwd:
+            managed_executable = os.path.join(cwd, "managed", "azcopy.exe")
+            cwd_executable = os.path.join(cwd, "azcopy.exe")
+
+            with mock.patch.object(util, "_get_default_install_location", return_value=managed_executable), \
+                    mock.patch.object(util.os, "getcwd", return_value=cwd), \
+                    mock.patch.object(util.shutil, "which", return_value=cwd_executable), \
+                    mock.patch.object(util.subprocess, "check_output") as check_output, \
+                    mock.patch.object(util.AzCopy, "install_azcopy") as install_azcopy:
+                azcopy = util.AzCopy()
+
+        check_output.assert_not_called()
+        install_azcopy.assert_called_once_with(managed_executable)
+        self.assertEqual(managed_executable, azcopy.executable)
+
+    def test_system_executable_is_probed_and_reused_by_absolute_path(self):
+        with tempfile.TemporaryDirectory() as cwd, tempfile.TemporaryDirectory() as system_dir:
+            managed_executable = os.path.join(cwd, "managed", "azcopy.exe")
+            system_executable = os.path.join(system_dir, "azcopy.exe")
+
+            with mock.patch.object(util, "_get_default_install_location", return_value=managed_executable), \
+                    mock.patch.object(util.os, "getcwd", return_value=cwd), \
+                    mock.patch.object(util.shutil, "which", return_value=system_executable), \
+                    mock.patch.object(util.subprocess, "check_output",
+                                      return_value=b"azcopy version 10.13.0\n") as check_output, \
+                    mock.patch.object(util.subprocess, "call", return_value=0) as call, \
+                    mock.patch.object(util.AzCopy, "install_azcopy") as install_azcopy:
+                azcopy = util.AzCopy()
+                azcopy.copy("source", "destination")
+
+        check_output.assert_called_once_with([os.path.abspath(system_executable), "--version"])
+        call.assert_called_once_with(
+            [os.path.abspath(system_executable), "copy", "source", "destination"],
+            env=mock.ANY)
+        install_azcopy.assert_not_called()
+        self.assertEqual(os.path.abspath(system_executable), azcopy.executable)
+
+    def test_relative_managed_executable_is_normalized_for_probe_and_operation(self):
+        relative_executable = os.path.join("managed", "azcopy.exe")
+        absolute_executable = os.path.abspath(relative_executable)
+
+        with mock.patch.object(util, "_get_default_install_location", return_value=relative_executable), \
+                mock.patch.object(util.os.path, "isfile", return_value=True), \
+                mock.patch.object(util.subprocess, "check_output",
+                                  return_value=b"azcopy version 10.13.0\n") as check_output, \
+                mock.patch.object(util.subprocess, "call", return_value=0) as call:
+            azcopy = util.AzCopy()
+            azcopy.sync("source", "destination")
+
+        check_output.assert_called_once_with([absolute_executable, "--version"])
+        call.assert_called_once_with(
+            [absolute_executable, "sync", "source", "destination"],
+            env=mock.ANY)
+        self.assertEqual(absolute_executable, azcopy.executable)
+
+    def test_system_probe_failure_falls_back_to_managed_executable(self):
+        with tempfile.TemporaryDirectory() as cwd, tempfile.TemporaryDirectory() as system_dir:
+            managed_executable = os.path.join(cwd, "managed", "azcopy.exe")
+            system_executable = os.path.join(system_dir, "azcopy.exe")
+
+            with mock.patch.object(util, "_get_default_install_location", return_value=managed_executable), \
+                    mock.patch.object(util.os, "getcwd", return_value=cwd), \
+                    mock.patch.object(util.shutil, "which", return_value=system_executable), \
+                    mock.patch.object(util.subprocess, "check_output", side_effect=OSError), \
+                    mock.patch.object(util.AzCopy, "install_azcopy") as install_azcopy:
+                azcopy = util.AzCopy()
+
+        install_azcopy.assert_called_once_with(managed_executable)
+        self.assertEqual(managed_executable, azcopy.executable)
+
+    def test_path_validation_failure_falls_back_without_probing(self):
+        managed_executable = os.path.abspath(os.path.join("managed", "azcopy.exe"))
+
+        with mock.patch.object(util, "_get_default_install_location", return_value=managed_executable), \
+                mock.patch.object(util.shutil, "which", return_value="azcopy.exe"), \
+                mock.patch.object(util.os, "getcwd", side_effect=OSError), \
+                mock.patch.object(util.subprocess, "check_output") as check_output, \
+                mock.patch.object(util.AzCopy, "install_azcopy") as install_azcopy:
+            azcopy = util.AzCopy()
+
+        check_output.assert_not_called()
+        install_azcopy.assert_called_once_with(managed_executable)
+        self.assertEqual(managed_executable, azcopy.executable)
+
+
+if __name__ == "__main__":
+    unittest.main()