[Storage] Fix #32852: az storage copy: Fix Azcopy download link to be using github release links

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

@@ -7,7 +7,6 @@
 import platform
 import subprocess
 import datetime
-import zipfile
 import stat
 from urllib.parse import urlparse
 from urllib.request import urlopen
@@ -56,36 +55,41 @@ def install_azcopy(self, install_location):
         install_dir = os.path.dirname(install_location)
         if not os.path.exists(install_dir):
             os.makedirs(install_dir)
-        base_url = 'https://azcopyvnext-awgzd8g7aagqhzhe.b02.azurefd.net/release20211027/azcopy_{}_{}_{}.{}'
+        file_extension = 'zip'
         if self.system == 'Windows':
             if platform.machine().endswith('64'):
-                file_url = base_url.format('windows', 'amd64', AZCOPY_VERSION, 'zip')
+                file_url = 'https://aka.ms/downloadazcopy-v10-windows'
                 if _verify_url(file_url) is None:
                     file_url = _verify_url('https://aka.ms/InstallAzCopyForCLIWindowsX64')
             else:
-                file_url = base_url.format('windows', '386', AZCOPY_VERSION, 'zip')
+                file_url = 'https://aka.ms/downloadazcopy-v10-windows-32bit'
                 if _verify_url(file_url) is None:
                     file_url = _verify_url('https://aka.ms/InstallAzCopyForCLIWindows')
         elif self.system == 'Linux':
-            file_url = base_url.format('linux', 'amd64', AZCOPY_VERSION, 'tar.gz')
+            file_extension = 'tar.gz'
+            file_url = 'https://aka.ms/downloadazcopy-v10-linux'
             if _verify_url(file_url) is None:
                 file_url = _verify_url('https://aka.ms/InstallAzCopyForCLILinux')
         elif self.system == 'Darwin':
-            file_url = base_url.format('darwin', 'amd64', AZCOPY_VERSION, 'zip')
+            file_url = 'https://aka.ms/downloadazcopy-v10-mac'
             if _verify_url(file_url) is None:
                 file_url = _verify_url('https://aka.ms/InstallAzCopyForCLIDarwin')
         else:
-            raise CLIError('Azcopy ({}) does not exist.'.format(self.system))
+            raise CLIError('Azcopy executable does not exist for {}.'.format(self.system))
+        azcopy_install_guide = 'https://learn.microsoft.com/azure/storage/common/storage-use-azcopy-v10'
+        if not file_url:
+            raise CLIError('Error while attempting to download azcopy. You could manually install the azcopy '
+                           'executable to {} by following the guide here: {}'.format(install_dir,
+                                                                                     azcopy_install_guide))
         try:
             os.chmod(install_dir, os.stat(install_dir).st_mode | stat.S_IWUSR)
-            _urlretrieve(file_url, install_location)
+            _urlretrieve(file_url, install_location, file_extension)
             os.chmod(install_location,
                      os.stat(install_location).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
-        except OSError as err:
-            azcopy_install_guide = 'https://learn.microsoft.com/azure/storage/common/storage-use-azcopy-v10'
-            raise CLIError('Connection error while attempting to download azcopy {}. You could also install the '
-                           'specified azcopy version to {} manually following the guide here: {} '
-                           '({})'.format(AZCOPY_VERSION, install_dir, azcopy_install_guide, err))
+        except (OSError, CLIError) as err:
+            raise CLIError('Error while attempting to download azcopy from {}. You could manually install the azcopy '
+                           'executable to {} by following the guide here: {} '
+                           '({})'.format(file_url, install_dir, azcopy_install_guide, err))
 
     def check_version(self):
         try:
@@ -226,34 +230,41 @@ def _get_default_install_location():
     return install_location
 
 
-def _urlretrieve(url, install_location):
+def _urlretrieve(url, install_location, file_extension):
     import io
     logger.warning('Downloading AzCopy from %s', url)
-    req = urlopen(url)
-    compressedFile = io.BytesIO(req.read())
-    if url.endswith('zip'):
-        zip_file = zipfile.ZipFile(compressedFile)
-        for fileName in zip_file.namelist():
-            if fileName.endswith('azcopy') or fileName.endswith('azcopy.exe'):
-                with open(install_location, 'wb') as f:
-                    f.write(zip_file.read(fileName))
-    elif url.endswith('gz'):
-        import tarfile
-        with tarfile.open(fileobj=compressedFile, mode="r:gz") as tar:
-            for tarinfo in tar:
-                if tarinfo.isfile() and tarinfo.name.endswith('azcopy'):
+    res = urlopen(url)
+    if res.status != 200:
+        raise CLIError('Invalid downloading url {}'.format(url))
+    compressedFile = io.BytesIO(res.read())
+    if file_extension == 'zip':
+        try:
+            import zipfile
+            zip_file = zipfile.ZipFile(compressedFile)
+            for fileName in zip_file.namelist():
+                if fileName.endswith('azcopy') or fileName.endswith('azcopy.exe'):
                     with open(install_location, 'wb') as f:
-                        f.write(tar.extractfile(tarinfo).read())
+                        f.write(zip_file.read(fileName))
+        except Exception as ex: # pylint: disable=broad-except
+            raise CLIError(ex)
     else:
-        raise CLIError('Invalid downloading url {}'.format(url))
+        try:
+            import tarfile
+            with tarfile.open(fileobj=compressedFile, mode="r:gz") as tar:
+                for tarinfo in tar:
+                    if tarinfo.isfile() and tarinfo.name.endswith('azcopy'):
+                        with open(install_location, 'wb') as f:
+                            f.write(tar.extractfile(tarinfo).read())
+        except Exception as ex: # pylint: disable=broad-except
+            raise CLIError(ex)
 
 
 def _verify_url(url):
     from urllib.error import HTTPError, URLError
     try:
         response = urlopen(url)
         if response.code == 200:
-            return response.url
+            return url
         return None
     except (HTTPError, URLError):
         logger.warning('There is an error downloading from the url: %s', url)