From 648e256eb11541531d375d86b063a80bb738cb72 Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Fri, 6 Feb 2026 16:40:51 +0530
Subject: [PATCH 01/12] feat(vm): add 'az vm cp' command for file transfer via
 storage bridge

Implements a client-side bridge solution using Azure Storage to facilitate file transfers to and from Azure VMs, particularly useful for private network environments. Leverages existing 'run-command' capabilities.
---
 .../azure/cli/command_modules/vm/_help.py     |  15 ++
 .../azure/cli/command_modules/vm/_params.py   |   6 +
 .../azure/cli/command_modules/vm/commands.py  |   1 +
 .../azure/cli/command_modules/vm/custom.py    | 179 ++++++++++++++++++
 4 files changed, 201 insertions(+)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/_help.py b/src/azure-cli/azure/cli/command_modules/vm/_help.py
index 60adcbc4d20..17687f5849f 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/_help.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/_help.py
@@ -987,6 +987,21 @@
     text: az vm list-sizes -l westus
 """
 
+helps['vm cp'] = """
+type: command
+short-summary: Copy files to and from a virtual machine.
+long-summary: >
+    This command uses an Azure Storage blob container as an intermediary bridge to transfer files.
+    It requires 'az vm run-command' capability on the target VM.
+examples:
+  - name: Upload a local file to a VM.
+    text: az vm cp --source /path/to/local/file --destination my-rg:my-vm:/path/to/remote/file
+  - name: Download a file from a VM to local.
+    text: az vm cp --source my-rg:my-vm:/path/to/remote/file --destination /path/to/local/file
+  - name: Upload a local file to a VM using a specific storage account.
+    text: az vm cp --source /path/to/local/file --destination my-vm:/path/to/remote/file --storage-account mystorageaccount
+"""
+
 helps['vm availability-set create'] = """
 type: command
 short-summary: Create an Azure Availability Set.
diff --git a/src/azure-cli/azure/cli/command_modules/vm/_params.py b/src/azure-cli/azure/cli/command_modules/vm/_params.py
index 8d6e77a6af2..90d301b5449 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/_params.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/_params.py
@@ -516,6 +516,12 @@ def load_arguments(self, _):
         with self.argument_context(scope) as c:
             c.ignore('include_user_data')
 
+    with self.argument_context('vm cp') as c:
+        c.argument('source', help='The source path. Use [resource-group:]vm-name:path for VM files, or a local path.')
+        c.argument('destination', help='The destination path. Use [resource-group:]vm-name:path for VM files, or a local path.')
+        c.argument('storage_account', help='The name or ID of the storage account to use as a bridge.')
+        c.argument('container_name', help='The name of the container to use in the storage account. Default: azvmcp')
+
     with self.argument_context('vm diagnostics') as c:
         c.argument('vm_name', arg_type=existing_vm_name, options_list=['--vm-name'])
 
diff --git a/src/azure-cli/azure/cli/command_modules/vm/commands.py b/src/azure-cli/azure/cli/command_modules/vm/commands.py
index 096df746e92..1385ac32a52 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/commands.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/commands.py
@@ -300,6 +300,7 @@ def load_command_table(self, _):
         g.wait_command('wait', getter_name='get_instance_view', getter_type=compute_custom)
         g.custom_command('auto-shutdown', 'auto_shutdown_vm')
         g.custom_command('list-sizes', 'list_vm_sizes', deprecate_info=g.deprecate(redirect='az vm list-skus'))
+        g.custom_command('cp', 'vm_cp')
 
         from .operations.vm import VMCapture
         self.command_table['vm capture'] = VMCapture(loader=self)
diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index 092820ada97..fc6c7632dd9 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -40,6 +40,9 @@
 from ._vm_utils import read_content_if_is_file, import_aaz_by_profile, IdentityType
 from ._vm_diagnostics_templates import get_default_diag_config
 
+from azure.cli.command_modules.storage.operations.blob import upload_blob, download_blob
+from azure.cli.command_modules.storage.util import create_short_lived_blob_sas_v2
+
 from ._actions import (load_images_from_aliases_doc, load_extension_images_thru_services,
                        load_images_thru_services, _get_latest_image_version, _get_latest_image_version_by_aaz)
 from ._client_factory import (_compute_client_factory, cf_vm_image_term)
@@ -6536,3 +6539,179 @@ def list_vm_sizes(cmd, location):
 
 
 # endRegion
+
+def _parse_vm_file_path(path):
+    if ':' not in path:
+        return None
+    
+    parts = path.split(':')
+    if len(parts) == 2:
+        # vm-name:path
+        return None, parts[0], parts[1]
+    if len(parts) == 3:
+        # rg:vm-name:path
+        return parts[0], parts[1], parts[2]
+    return None
+
+
+def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp'):
+    source_vm = _parse_vm_file_path(source)
+    dest_vm = _parse_vm_file_path(destination)
+    
+    if source_vm and dest_vm:
+        raise ValidationError("Both source and destination cannot be VM paths.")
+    if not source_vm and not dest_vm:
+        raise ValidationError("Either source or destination must be a VM path (format: [rg:]vm:path).")
+    
+    # 1. Prepare Storage Account
+    if not storage_account:
+        # Try to find a storage account in the VM's resource group
+        rg = (source_vm[0] if source_vm else dest_vm[0])
+        vm_name = (source_vm[1] if source_vm else dest_vm[1])
+        
+        if not rg:
+            # Get RG of the VM
+            client = _compute_client_factory(cmd.cli_ctx)
+            vms = client.virtual_machines.list_all()
+            vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
+            if not vm:
+                raise ResourceNotFoundError("VM '{}' not found.".format(vm_name))
+            # parse RG from ID
+            rg = vm.id.split('/')[4]
+        
+        from azure.cli.command_modules.storage._client_factory import cf_sa
+        sa_client = cf_sa(cmd.cli_ctx, None)
+        accounts = list(sa_client.list())
+        # Filter by RG if possible
+        rg_accounts = [a for a in accounts if a.id.split('/')[4].lower() == rg.lower()]
+        if rg_accounts:
+            storage_account = rg_accounts[0].name
+        elif accounts:
+            storage_account = accounts[0].name
+        else:
+            raise RequiredArgumentMissingError("No storage account found in the subscription. Please provide one with --storage-account.")
+
+    # Get account key
+    from azure.cli.command_modules.storage._client_factory import cf_sa_for_keys
+    sa_keys_client = cf_sa_for_keys(cmd.cli_ctx, None)
+    # Check if storage_account is name or ID
+    if '/' in storage_account:
+        sa_rg = storage_account.split('/')[4]
+        sa_name = storage_account.split('/')[-1]
+    else:
+        # Search for it
+        sa_client = cf_sa(cmd.cli_ctx, None)
+        accounts = list(sa_client.list())
+        account = next((a for a in accounts if a.name.lower() == storage_account.lower()), None)
+        if not account:
+            raise ResourceNotFoundError("Storage account '{}' not found.".format(storage_account))
+        sa_rg = account.id.split('/')[4]
+        sa_name = account.name
+    
+    keys = sa_keys_client.list_keys(sa_rg, sa_name).keys
+    account_key = keys[0].value
+
+    # Ensure container exists
+    from azure.cli.command_modules.storage._client_factory import cf_blob_service
+    blob_service_client = cf_blob_service(cmd.cli_ctx, {'account_name': sa_name, 'account_key': account_key})
+    container_client = blob_service_client.get_container_client(container_name)
+    try:
+        container_client.create_container()
+    except Exception: # Already exists or other error
+        pass
+
+    import uuid
+    blob_name = str(uuid.uuid4())
+
+    blob_client = container_client.get_blob_client(blob_name)
+
+    if dest_vm:
+        # UPLOAD: Local -> VM
+        rg, vm_name, vm_path = dest_vm
+        if not rg:
+            # find VM RG
+            client = _compute_client_factory(cmd.cli_ctx)
+            vms = client.virtual_machines.list_all()
+            vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
+            rg = vm.id.split('/')[4]
+
+        logger.info("Uploading local file to bridge storage...")
+        upload_blob(cmd, blob_client, file_path=source)
+        
+        # Get SAS for VM to download
+        sas_token = create_short_lived_blob_sas_v2(cmd, sa_name, container_name, blob_name, account_key=account_key)
+        blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
+
+        # VM run-command to download
+        # Check OS type
+        vm_obj = _compute_client_factory(cmd.cli_ctx).virtual_machines.get(rg, vm_name)
+        is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
+        
+        if is_linux:
+            script = "curl -L -o '{}' '{}'".format(vm_path, blob_url)
+            command_id = 'RunShellScript'
+        else:
+            script = "Invoke-WebRequest -Uri '{}' -OutFile '{}'".format(blob_url, vm_path)
+            command_id = 'RunPowerShellScript'
+            
+        logger.info("Executing download script in VM...")
+        from .aaz.latest.vm.run_command import Invoke
+        Invoke(cli_ctx=cmd.cli_ctx)(command_args={
+            'resource_group': rg,
+            'vm_name': vm_name,
+            'command_id': command_id,
+            'script': [script]
+        })
+        
+        # Cleanup
+        logger.info("Cleaning up bridge storage...")
+        blob_client.delete_blob()
+        
+    else:
+        # DOWNLOAD: VM -> Local
+        rg, vm_name, vm_path = source_vm
+        if not rg:
+             # find VM RG
+            client = _compute_client_factory(cmd.cli_ctx)
+            vms = client.virtual_machines.list_all()
+            vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
+            rg = vm.id.split('/')[4]
+
+        # Get SAS with WRITE permission
+        from datetime import datetime, timedelta
+        from azure.cli.core.profiles import ResourceType
+        t_sas = cmd.get_models('_shared_access_signature#BlobSharedAccessSignature',
+                               resource_type=ResourceType.DATA_STORAGE_BLOB)
+        t_blob_permissions = cmd.get_models('_models#BlobSasPermissions', resource_type=ResourceType.DATA_STORAGE_BLOB)
+        expiry = (datetime.utcnow() + timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%SZ')
+        sas = t_sas(sa_name, account_key=account_key)
+        sas_token = sas.generate_blob(container_name, blob_name, permission=t_blob_permissions(write=True), expiry=expiry, protocol='https')
+        blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
+
+        vm_obj = _compute_client_factory(cmd.cli_ctx).virtual_machines.get(rg, vm_name)
+        is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
+
+        if is_linux:
+            script = "curl -X PUT -T '{}' -H 'x-ms-blob-type: BlockBlob' '{}'".format(vm_path, blob_url)
+            command_id = 'RunShellScript'
+        else:
+            script = "$body = Get-Content -Path '{}' -Encoding Byte; Invoke-RestMethod -Uri '{}' -Method Put -Headers @{{'x-ms-blob-type'='BlockBlob'}} -Body $body".format(vm_path, blob_url)
+            command_id = 'RunPowerShellScript'
+
+        logger.info("Executing upload script in VM...")
+        from .aaz.latest.vm.run_command import Invoke
+        Invoke(cli_ctx=cmd.cli_ctx)(command_args={
+            'resource_group': rg,
+            'vm_name': vm_name,
+            'command_id': command_id,
+            'script': [script]
+        })
+
+        logger.info("Downloading from bridge storage to local...")
+        download_blob(blob_client, file_path=destination)
+
+        # Cleanup
+        logger.info("Cleaning up bridge storage...")
+        blob_client.delete_blob()
+
+    return {"message": "File transfer successful."}

From 51903cd64a778743bdec61391afad7ebe114a794 Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Fri, 6 Feb 2026 16:42:59 +0530
Subject: [PATCH 02/12] test(vm): add unit tests for 'az vm cp' path parsing
 and upload flow

---
 .../vm/tests/latest/test_vm_cp_unit.py        | 69 +++++++++++++++++++
 1 file changed, 69 insertions(+)
 create mode 100644 src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py

diff --git a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
new file mode 100644
index 00000000000..4588f27a06f
--- /dev/null
+++ b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
@@ -0,0 +1,69 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import unittest
+from unittest import mock
+from azure.cli.command_modules.vm.custom import _parse_vm_file_path, vm_cp
+
+class TestVmCp(unittest.TestCase):
+
+    def test_parse_vm_file_path(self):
+        # Local path
+        self.assertIsNone(_parse_vm_file_path("/path/to/file"))
+        self.assertIsNone(_parse_vm_file_path("C:\\path\\to\\file"))
+        
+        # VM path: vm:path
+        self.assertEqual(_parse_vm_file_path("myvm:/tmp/file"), (None, "myvm", "/tmp/file"))
+        
+        # VM path: rg:vm:path
+        self.assertEqual(_parse_vm_file_path("myrg:myvm:/tmp/file"), ("myrg", "myvm", "/tmp/file"))
+        
+        # Edge cases
+        self.assertIsNone(_parse_vm_file_path("justfile"))
+        self.assertEqual(_parse_vm_file_path("vm:C:\\path"), (None, "vm", "C:\\path"))
+        self.assertEqual(_parse_vm_file_path("rg:vm:C:\\path"), ("rg", "vm", "C:\\path"))
+
+    @mock.patch('azure.cli.command_modules.vm.custom._compute_client_factory')
+    @mock.patch('azure.cli.command_modules.vm.custom.get_storage_client_factory')
+    @mock.patch('azure.cli.command_modules.vm.custom.create_short_lived_blob_sas_v2')
+    @mock.patch('azure.cli.command_modules.vm.custom.upload_blob')
+    def test_vm_cp_upload_basic(self, mock_upload, mock_sas, mock_storage_factory, mock_compute_factory):
+        cmd = mock.MagicMock()
+        cmd.cli_ctx.cloud.suffixes.storage_endpoint = 'core.windows.net'
+        
+        # Mock compute client
+        mock_compute = mock.MagicMock()
+        mock_compute_factory.return_value = mock_compute
+        
+        vm_obj = mock.MagicMock()
+        vm_obj.storage_profile.os_disk.os_type.lower.return_value = 'linux'
+        mock_compute.virtual_machines.get.return_value = vm_obj
+        
+        # Mock storage client
+        mock_storage = mock.MagicMock()
+        mock_storage_factory.return_value = mock_storage
+        
+        sa = mock.MagicMock()
+        sa.name = 'mystorage'
+        mock_storage.storage_accounts.list_by_resource_group.return_value = [sa]
+        
+        key = mock.MagicMock()
+        key.value = 'key1'
+        mock_storage.storage_accounts.list_keys.return_value.keys = [key]
+        
+        # Mock blob client
+        with mock.patch('azure.cli.command_modules.vm.custom.BlobServiceClient') as mock_blob_service:
+            mock_container = mock.MagicMock()
+            mock_blob_service.from_connection_string.return_value.get_container_client.return_value = mock_container
+            
+            # Execute
+            vm_cp(cmd, source="local.txt", destination="myrg:myvm:/tmp/remote.txt")
+            
+            # Verify
+            mock_upload.assert_called_once()
+            mock_compute.virtual_machines.get.assert_called_with("myrg", "myvm")
+
+if __name__ == '__main__':
+    unittest.main()

From eaf303d2e5167f3960d6556190be8ca4e1376e30 Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Fri, 6 Feb 2026 16:51:19 +0530
Subject: [PATCH 03/12] style(vm): cleanup imports and formatting in 'az vm cp'
 implementation

---
 .../azure/cli/command_modules/vm/custom.py    | 33 ++++++++++---------
 1 file changed, 17 insertions(+), 16 deletions(-)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index fc6c7632dd9..79de079ca28 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -14,6 +14,8 @@
 # pylint: disable=protected-access
 import json
 import os
+import uuid
+from datetime import datetime, timedelta
 
 import requests
 
@@ -6543,7 +6545,7 @@ def list_vm_sizes(cmd, location):
 def _parse_vm_file_path(path):
     if ':' not in path:
         return None
-    
+
     parts = path.split(':')
     if len(parts) == 2:
         # vm-name:path
@@ -6557,18 +6559,18 @@ def _parse_vm_file_path(path):
 def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp'):
     source_vm = _parse_vm_file_path(source)
     dest_vm = _parse_vm_file_path(destination)
-    
+
     if source_vm and dest_vm:
         raise ValidationError("Both source and destination cannot be VM paths.")
     if not source_vm and not dest_vm:
         raise ValidationError("Either source or destination must be a VM path (format: [rg:]vm:path).")
-    
+
     # 1. Prepare Storage Account
     if not storage_account:
         # Try to find a storage account in the VM's resource group
         rg = (source_vm[0] if source_vm else dest_vm[0])
         vm_name = (source_vm[1] if source_vm else dest_vm[1])
-        
+
         if not rg:
             # Get RG of the VM
             client = _compute_client_factory(cmd.cli_ctx)
@@ -6578,7 +6580,7 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
                 raise ResourceNotFoundError("VM '{}' not found.".format(vm_name))
             # parse RG from ID
             rg = vm.id.split('/')[4]
-        
+
         from azure.cli.command_modules.storage._client_factory import cf_sa
         sa_client = cf_sa(cmd.cli_ctx, None)
         accounts = list(sa_client.list())
@@ -6607,7 +6609,7 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
             raise ResourceNotFoundError("Storage account '{}' not found.".format(storage_account))
         sa_rg = account.id.split('/')[4]
         sa_name = account.name
-    
+
     keys = sa_keys_client.list_keys(sa_rg, sa_name).keys
     account_key = keys[0].value
 
@@ -6617,10 +6619,9 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
     container_client = blob_service_client.get_container_client(container_name)
     try:
         container_client.create_container()
-    except Exception: # Already exists or other error
+    except Exception: # Already exists or other error  # pylint: disable=broad-except
         pass
 
-    import uuid
     blob_name = str(uuid.uuid4())
 
     blob_client = container_client.get_blob_client(blob_name)
@@ -6637,7 +6638,7 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
 
         logger.info("Uploading local file to bridge storage...")
         upload_blob(cmd, blob_client, file_path=source)
-        
+
         # Get SAS for VM to download
         sas_token = create_short_lived_blob_sas_v2(cmd, sa_name, container_name, blob_name, account_key=account_key)
         blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
@@ -6646,14 +6647,14 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
         # Check OS type
         vm_obj = _compute_client_factory(cmd.cli_ctx).virtual_machines.get(rg, vm_name)
         is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
-        
+
         if is_linux:
             script = "curl -L -o '{}' '{}'".format(vm_path, blob_url)
             command_id = 'RunShellScript'
         else:
             script = "Invoke-WebRequest -Uri '{}' -OutFile '{}'".format(blob_url, vm_path)
             command_id = 'RunPowerShellScript'
-            
+
         logger.info("Executing download script in VM...")
         from .aaz.latest.vm.run_command import Invoke
         Invoke(cli_ctx=cmd.cli_ctx)(command_args={
@@ -6662,11 +6663,11 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
             'command_id': command_id,
             'script': [script]
         })
-        
+
         # Cleanup
         logger.info("Cleaning up bridge storage...")
         blob_client.delete_blob()
-        
+
     else:
         # DOWNLOAD: VM -> Local
         rg, vm_name, vm_path = source_vm
@@ -6678,14 +6679,14 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
             rg = vm.id.split('/')[4]
 
         # Get SAS with WRITE permission
-        from datetime import datetime, timedelta
-        from azure.cli.core.profiles import ResourceType
         t_sas = cmd.get_models('_shared_access_signature#BlobSharedAccessSignature',
                                resource_type=ResourceType.DATA_STORAGE_BLOB)
         t_blob_permissions = cmd.get_models('_models#BlobSasPermissions', resource_type=ResourceType.DATA_STORAGE_BLOB)
         expiry = (datetime.utcnow() + timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%SZ')
         sas = t_sas(sa_name, account_key=account_key)
-        sas_token = sas.generate_blob(container_name, blob_name, permission=t_blob_permissions(write=True), expiry=expiry, protocol='https')
+        sas_token = sas.generate_blob(container_name, blob_name,
+                                      permission=t_blob_permissions(write=True),
+                                      expiry=expiry, protocol='https')
         blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
 
         vm_obj = _compute_client_factory(cmd.cli_ctx).virtual_machines.get(rg, vm_name)

From 6203d555e7827b3991d3ac5a28218dd477fb8d42 Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Fri, 6 Feb 2026 16:56:56 +0530
Subject: [PATCH 04/12] fix(vm): address security, reliability and path parsing
 feedback for 'az vm cp'

---
 .../azure/cli/command_modules/vm/custom.py    | 240 +++++++++++-------
 .../vm/tests/latest/test_vm_cp_unit.py        |  54 ++--
 2 files changed, 181 insertions(+), 113 deletions(-)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index 79de079ca28..c5394287c49 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -6542,21 +6542,58 @@ def list_vm_sizes(cmd, location):
 
 # endRegion
 
+def _is_windows_absolute_path(path):
+    """
+    Detect a Windows-style absolute path such as 'C:\\path\\to\\file' or 'D:/file.txt'.
+    This is a purely syntactic check and does not touch the filesystem.
+    """
+    return (
+        isinstance(path, str) and
+        len(path) >= 3 and
+        path[0].isalpha() and
+        path[1] == ':' and
+        path[2] in ('\\', '/')
+    )
+
+
 def _parse_vm_file_path(path):
+    # If there is no colon, this cannot be a VM path.
     if ':' not in path:
         return None
 
-    parts = path.split(':')
-    if len(parts) == 2:
+    # Do not treat Windows absolute paths like 'C:\\path\\to\\file' as VM paths.
+    if _is_windows_absolute_path(path):
+        return None
+
+    # VM path format is [resource-group-name:]vm-name:path-on-vm
+    # Only the first two colons separate components; any remaining colons belong to the path.
+    first_colon = path.find(':')
+    second_colon = path.find(':', first_colon + 1)
+
+    if first_colon == -1:
+        return None
+
+    if second_colon == -1:
         # vm-name:path
-        return None, parts[0], parts[1]
-    if len(parts) == 3:
-        # rg:vm-name:path
-        return parts[0], parts[1], parts[2]
-    return None
+        vm_name = path[:first_colon]
+        vm_path = path[first_colon + 1:]
+        if not vm_name or not vm_path:
+            return None
+        return None, vm_name, vm_path
+
+    # rg:vm-name:path...
+    rg_name = path[:first_colon]
+    vm_name = path[first_colon + 1:second_colon]
+    vm_path = path[second_colon + 1:]
+    if not rg_name or not vm_name or not vm_path:
+        return None
+    return rg_name, vm_name, vm_path
 
 
 def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp'):
+    from azure.core.exceptions import ResourceExistsError
+    import shlex
+
     source_vm = _parse_vm_file_path(source)
     dest_vm = _parse_vm_file_path(destination)
 
@@ -6619,100 +6656,117 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
     container_client = blob_service_client.get_container_client(container_name)
     try:
         container_client.create_container()
-    except Exception: # Already exists or other error  # pylint: disable=broad-except
+    except ResourceExistsError:
         pass
 
     blob_name = str(uuid.uuid4())
-
     blob_client = container_client.get_blob_client(blob_name)
 
-    if dest_vm:
-        # UPLOAD: Local -> VM
-        rg, vm_name, vm_path = dest_vm
-        if not rg:
-            # find VM RG
-            client = _compute_client_factory(cmd.cli_ctx)
-            vms = client.virtual_machines.list_all()
-            vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
-            rg = vm.id.split('/')[4]
-
-        logger.info("Uploading local file to bridge storage...")
-        upload_blob(cmd, blob_client, file_path=source)
-
-        # Get SAS for VM to download
-        sas_token = create_short_lived_blob_sas_v2(cmd, sa_name, container_name, blob_name, account_key=account_key)
-        blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
-
-        # VM run-command to download
-        # Check OS type
-        vm_obj = _compute_client_factory(cmd.cli_ctx).virtual_machines.get(rg, vm_name)
-        is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
-
-        if is_linux:
-            script = "curl -L -o '{}' '{}'".format(vm_path, blob_url)
-            command_id = 'RunShellScript'
-        else:
-            script = "Invoke-WebRequest -Uri '{}' -OutFile '{}'".format(blob_url, vm_path)
-            command_id = 'RunPowerShellScript'
-
-        logger.info("Executing download script in VM...")
-        from .aaz.latest.vm.run_command import Invoke
-        Invoke(cli_ctx=cmd.cli_ctx)(command_args={
-            'resource_group': rg,
-            'vm_name': vm_name,
-            'command_id': command_id,
-            'script': [script]
-        })
-
-        # Cleanup
-        logger.info("Cleaning up bridge storage...")
-        blob_client.delete_blob()
-
-    else:
-        # DOWNLOAD: VM -> Local
-        rg, vm_name, vm_path = source_vm
-        if not rg:
-             # find VM RG
-            client = _compute_client_factory(cmd.cli_ctx)
-            vms = client.virtual_machines.list_all()
-            vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
-            rg = vm.id.split('/')[4]
+    try:
+        if dest_vm:
+            # UPLOAD: Local -> VM
+            rg, vm_name, vm_path = dest_vm
+            if not rg:
+                # find VM RG
+                client = _compute_client_factory(cmd.cli_ctx)
+                vms = client.virtual_machines.list_all()
+                vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
+                rg = vm.id.split('/')[4]
+
+            logger.info("Uploading local file to bridge storage...")
+            upload_blob(cmd, blob_client, file_path=source)
+
+            # Get SAS for VM to download (2 hours expiry)
+            sas_token = create_short_lived_blob_sas_v2(cmd, sa_name, container_name, blob_name, account_key=account_key)
+            blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
+
+            # VM run-command to download
+            # Check OS type
+            vm_obj = _compute_client_factory(cmd.cli_ctx).virtual_machines.get(rg, vm_name)
+            is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
+
+            if is_linux:
+                script = "curl -L -f -s -S -o {} {}".format(shlex.quote(vm_path), shlex.quote(blob_url))
+                command_id = 'RunShellScript'
+            else:
+                # Escape single quotes for PowerShell
+                escaped_vm_path = vm_path.replace("'", "''")
+                escaped_blob_url = blob_url.replace("'", "''")
+                script = "Invoke-WebRequest -Uri '{}' -OutFile '{}'".format(escaped_blob_url, escaped_vm_path)
+                command_id = 'RunPowerShellScript'
+
+            logger.info("Executing download script in VM...")
+            from .aaz.latest.vm.run_command import Invoke
+            result = Invoke(cli_ctx=cmd.cli_ctx)(command_args={
+                'resource_group': rg,
+                'vm_name': vm_name,
+                'command_id': command_id,
+                'script': [script]
+            })
+            if result.get('value') and result['value'][0].get('message'):
+                message = result['value'][0]['message']
+                if 'failed' in message.lower() or 'error' in message.lower():
+                    raise CLIError("VM execution failed: {}".format(message))
 
-        # Get SAS with WRITE permission
-        t_sas = cmd.get_models('_shared_access_signature#BlobSharedAccessSignature',
-                               resource_type=ResourceType.DATA_STORAGE_BLOB)
-        t_blob_permissions = cmd.get_models('_models#BlobSasPermissions', resource_type=ResourceType.DATA_STORAGE_BLOB)
-        expiry = (datetime.utcnow() + timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%SZ')
-        sas = t_sas(sa_name, account_key=account_key)
-        sas_token = sas.generate_blob(container_name, blob_name,
-                                      permission=t_blob_permissions(write=True),
-                                      expiry=expiry, protocol='https')
-        blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
-
-        vm_obj = _compute_client_factory(cmd.cli_ctx).virtual_machines.get(rg, vm_name)
-        is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
-
-        if is_linux:
-            script = "curl -X PUT -T '{}' -H 'x-ms-blob-type: BlockBlob' '{}'".format(vm_path, blob_url)
-            command_id = 'RunShellScript'
         else:
-            script = "$body = Get-Content -Path '{}' -Encoding Byte; Invoke-RestMethod -Uri '{}' -Method Put -Headers @{{'x-ms-blob-type'='BlockBlob'}} -Body $body".format(vm_path, blob_url)
-            command_id = 'RunPowerShellScript'
-
-        logger.info("Executing upload script in VM...")
-        from .aaz.latest.vm.run_command import Invoke
-        Invoke(cli_ctx=cmd.cli_ctx)(command_args={
-            'resource_group': rg,
-            'vm_name': vm_name,
-            'command_id': command_id,
-            'script': [script]
-        })
+            # DOWNLOAD: VM -> Local
+            rg, vm_name, vm_path = source_vm
+            if not rg:
+                # find VM RG
+                client = _compute_client_factory(cmd.cli_ctx)
+                vms = client.virtual_machines.list_all()
+                vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
+                rg = vm.id.split('/')[4]
+
+            # Get SAS with WRITE permission (2 hours expiry)
+            t_sas = cmd.get_models('_shared_access_signature#BlobSharedAccessSignature',
+                                   resource_type=ResourceType.DATA_STORAGE_BLOB)
+            t_blob_permissions = cmd.get_models('_models#BlobSasPermissions', resource_type=ResourceType.DATA_STORAGE_BLOB)
+            expiry = (datetime.utcnow() + timedelta(hours=2)).strftime('%Y-%m-%dT%H:%M:%SZ')
+            sas = t_sas(sa_name, account_key=account_key)
+            sas_token = sas.generate_blob(container_name, blob_name,
+                                          permission=t_blob_permissions(write=True),
+                                          expiry=expiry, protocol='https')
+            blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
+
+            vm_obj = _compute_client_factory(cmd.cli_ctx).virtual_machines.get(rg, vm_name)
+            is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
+
+            if is_linux:
+                script = "curl -X PUT -T {} -H 'x-ms-blob-type: BlockBlob' {}".format(shlex.quote(vm_path), shlex.quote(blob_url))
+                command_id = 'RunShellScript'
+            else:
+                escaped_vm_path = vm_path.replace("'", "''")
+                escaped_blob_url = blob_url.replace("'", "''")
+                # Cross-version compatible Get-Content as suggested by Copilot
+                script = ("$body = if ($PSVersionTable.PSVersion.Major -ge 6) {{ Get-Content -Path '{}' -AsByteStream }} "
+                          "else {{ Get-Content -Path '{}' -Encoding Byte }}; "
+                          "Invoke-RestMethod -Uri '{}' -Method Put -Headers @{{'x-ms-blob-type'='BlockBlob'}} -Body $body").format(
+                              escaped_vm_path, escaped_vm_path, escaped_blob_url)
+                command_id = 'RunPowerShellScript'
+
+            logger.info("Executing upload script in VM...")
+            from .aaz.latest.vm.run_command import Invoke
+            result = Invoke(cli_ctx=cmd.cli_ctx)(command_args={
+                'resource_group': rg,
+                'vm_name': vm_name,
+                'command_id': command_id,
+                'script': [script]
+            })
+            if result.get('value') and result['value'][0].get('message'):
+                message = result['value'][0]['message']
+                if 'failed' in message.lower() or 'error' in message.lower():
+                    raise CLIError("VM execution failed: {}".format(message))
 
-        logger.info("Downloading from bridge storage to local...")
-        download_blob(blob_client, file_path=destination)
+            logger.info("Downloading from bridge storage to local...")
+            download_blob(blob_client, file_path=destination)
 
-        # Cleanup
-        logger.info("Cleaning up bridge storage...")
-        blob_client.delete_blob()
+    finally:
+        # Cleanup bridge storage
+        try:
+            logger.info("Cleaning up bridge storage...")
+            blob_client.delete_blob()
+        except Exception:  # pylint: disable=broad-except
+            pass
 
     return {"message": "File transfer successful."}
diff --git a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
index 4588f27a06f..b61689db6bc 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
@@ -10,9 +10,11 @@
 class TestVmCp(unittest.TestCase):
 
     def test_parse_vm_file_path(self):
-        # Local path
+        # Local paths (non-VM)
         self.assertIsNone(_parse_vm_file_path("/path/to/file"))
         self.assertIsNone(_parse_vm_file_path("C:\\path\\to\\file"))
+        self.assertIsNone(_parse_vm_file_path("D:/path/to/file"))
+        self.assertIsNone(_parse_vm_file_path("justfile"))
         
         # VM path: vm:path
         self.assertEqual(_parse_vm_file_path("myvm:/tmp/file"), (None, "myvm", "/tmp/file"))
@@ -20,16 +22,17 @@ def test_parse_vm_file_path(self):
         # VM path: rg:vm:path
         self.assertEqual(_parse_vm_file_path("myrg:myvm:/tmp/file"), ("myrg", "myvm", "/tmp/file"))
         
-        # Edge cases
-        self.assertIsNone(_parse_vm_file_path("justfile"))
-        self.assertEqual(_parse_vm_file_path("vm:C:\\path"), (None, "vm", "C:\\path"))
-        self.assertEqual(_parse_vm_file_path("rg:vm:C:\\path"), ("rg", "vm", "C:\\path"))
+        # VM path with colons in the path component
+        self.assertEqual(_parse_vm_file_path("vm:C:\\remote\\path"), (None, "vm", "C:\\remote\\path"))
+        self.assertEqual(_parse_vm_file_path("rg:vm:C:\\remote\\path"), ("rg", "vm", "C:\\remote\\path"))
 
     @mock.patch('azure.cli.command_modules.vm.custom._compute_client_factory')
-    @mock.patch('azure.cli.command_modules.vm.custom.get_storage_client_factory')
+    @mock.patch('azure.cli.command_modules.vm.custom.cf_sa')
+    @mock.patch('azure.cli.command_modules.vm.custom.cf_sa_for_keys')
+    @mock.patch('azure.cli.command_modules.vm.custom.cf_blob_service')
     @mock.patch('azure.cli.command_modules.vm.custom.create_short_lived_blob_sas_v2')
     @mock.patch('azure.cli.command_modules.vm.custom.upload_blob')
-    def test_vm_cp_upload_basic(self, mock_upload, mock_sas, mock_storage_factory, mock_compute_factory):
+    def test_vm_cp_upload_basic(self, mock_upload, mock_sas, mock_blob_factory, mock_keys_factory, mock_sa_factory, mock_compute_factory):
         cmd = mock.MagicMock()
         cmd.cli_ctx.cloud.suffixes.storage_endpoint = 'core.windows.net'
         
@@ -39,31 +42,42 @@ def test_vm_cp_upload_basic(self, mock_upload, mock_sas, mock_storage_factory, m
         
         vm_obj = mock.MagicMock()
         vm_obj.storage_profile.os_disk.os_type.lower.return_value = 'linux'
+        vm_obj.id = "/subscriptions/sub/resourceGroups/myrg/providers/Microsoft.Compute/virtualMachines/myvm"
         mock_compute.virtual_machines.get.return_value = vm_obj
+        mock_compute.virtual_machines.list_all.return_value = [vm_obj]
         
-        # Mock storage client
-        mock_storage = mock.MagicMock()
-        mock_storage_factory.return_value = mock_storage
+        # Mock storage clients
+        mock_sa = mock.MagicMock()
+        mock_sa_factory.return_value = mock_sa
         
         sa = mock.MagicMock()
         sa.name = 'mystorage'
-        mock_storage.storage_accounts.list_by_resource_group.return_value = [sa]
+        sa.id = "/subscriptions/sub/resourceGroups/myrg/providers/Microsoft.Storage/storageAccounts/mystorage"
+        mock_sa.list.return_value = [sa]
         
+        mock_keys = mock.MagicMock()
+        mock_keys_factory.return_value = mock_keys
         key = mock.MagicMock()
         key.value = 'key1'
-        mock_storage.storage_accounts.list_keys.return_value.keys = [key]
+        mock_keys.list_keys.return_value.keys = [key]
         
-        # Mock blob client
-        with mock.patch('azure.cli.command_modules.vm.custom.BlobServiceClient') as mock_blob_service:
-            mock_container = mock.MagicMock()
-            mock_blob_service.from_connection_string.return_value.get_container_client.return_value = mock_container
-            
-            # Execute
-            vm_cp(cmd, source="local.txt", destination="myrg:myvm:/tmp/remote.txt")
+        # Mock blob service
+        mock_blob_service = mock.MagicMock()
+        mock_blob_factory.return_value = mock_blob_service
+        mock_container = mock.MagicMock()
+        mock_blob_service.get_container_client.return_value = mock_container
+        mock_blob = mock.MagicMock()
+        mock_container.get_blob_client.return_value = mock_blob
+        
+        # Execute
+        with mock.patch('azure.cli.command_modules.vm.custom.Invoke') as mock_invoke:
+            mock_invoke.return_value.return_value = {'value': [{'message': 'success'}]}
+            vm_cp(cmd, source="local.txt", destination="myvm:/tmp/remote.txt")
             
             # Verify
             mock_upload.assert_called_once()
-            mock_compute.virtual_machines.get.assert_called_with("myrg", "myvm")
+            mock_compute.virtual_machines.get.assert_called()
+            mock_blob.delete_blob.assert_called_once()
 
 if __name__ == '__main__':
     unittest.main()

From a723e318ae3efa1cadfd62a0505126dd51387a0e Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Fri, 6 Feb 2026 17:02:16 +0530
Subject: [PATCH 05/12] style(vm): finalize imports and code structure for 'az
 vm cp'

---
 src/azure-cli/azure/cli/command_modules/vm/custom.py | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index c5394287c49..7ddf0f73419 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -15,6 +15,7 @@
 import json
 import os
 import uuid
+import shlex
 from datetime import datetime, timedelta
 
 import requests
@@ -24,6 +25,7 @@
 
 from knack.log import get_logger
 from knack.util import CLIError
+from azure.core.exceptions import ResourceExistsError
 from azure.cli.core.azclierror import (
     ResourceNotFoundError,
     ValidationError,
@@ -6591,8 +6593,7 @@ def _parse_vm_file_path(path):
 
 
 def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp'):
-    from azure.core.exceptions import ResourceExistsError
-    import shlex
+    from .aaz.latest.vm.run_command import Invoke
 
     source_vm = _parse_vm_file_path(source)
     dest_vm = _parse_vm_file_path(destination)
@@ -6696,7 +6697,6 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
                 command_id = 'RunPowerShellScript'
 
             logger.info("Executing download script in VM...")
-            from .aaz.latest.vm.run_command import Invoke
             result = Invoke(cli_ctx=cmd.cli_ctx)(command_args={
                 'resource_group': rg,
                 'vm_name': vm_name,
@@ -6746,7 +6746,6 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
                 command_id = 'RunPowerShellScript'
 
             logger.info("Executing upload script in VM...")
-            from .aaz.latest.vm.run_command import Invoke
             result = Invoke(cli_ctx=cmd.cli_ctx)(command_args={
                 'resource_group': rg,
                 'vm_name': vm_name,

From afebf60bbc227ab436beeb793241c0d716bd11fc Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Fri, 6 Feb 2026 18:30:20 +0530
Subject: [PATCH 06/12] fix(vm): fix duplicate argument, extract helper, and
 update doc

---
 .../azure/cli/command_modules/vm/_help.py     |  4 +-
 .../azure/cli/command_modules/vm/custom.py    | 56 +++++++++----------
 2 files changed, 29 insertions(+), 31 deletions(-)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/_help.py b/src/azure-cli/azure/cli/command_modules/vm/_help.py
index 17687f5849f..02daab29b8d 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/_help.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/_help.py
@@ -992,7 +992,9 @@
 short-summary: Copy files to and from a virtual machine.
 long-summary: >
     This command uses an Azure Storage blob container as an intermediary bridge to transfer files.
-    It requires 'az vm run-command' capability on the target VM.
+    It requires that the target VM is running and accessible via 'az vm run-command'. On Linux VMs, 'curl'
+    must be installed and available on the PATH. The identity used to run this command must have sufficient
+    permissions both on the VM (to execute run-command) and on the storage account/container (to read and write blobs).
 examples:
   - name: Upload a local file to a VM.
     text: az vm cp --source /path/to/local/file --destination my-rg:my-vm:/path/to/remote/file
diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index 7ddf0f73419..3823bb51cce 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -6592,6 +6592,22 @@ def _parse_vm_file_path(path):
     return rg_name, vm_name, vm_path
 
 
+def _get_vm_and_rg(cmd, vm_name, rg=None):
+    client = _compute_client_factory(cmd.cli_ctx)
+    if rg:
+        vm = client.virtual_machines.get(rg, vm_name)
+        return vm, rg
+
+    # Search for VM across all RGs
+    vms = client.virtual_machines.list_all()
+    vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
+    if not vm:
+        raise ResourceNotFoundError("VM '{}' not found.".format(vm_name))
+    # parse RG from ID
+    rg = vm.id.split('/')[4]
+    return vm, rg
+
+
 def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp'):
     from .aaz.latest.vm.run_command import Invoke
 
@@ -6606,18 +6622,10 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
     # 1. Prepare Storage Account
     if not storage_account:
         # Try to find a storage account in the VM's resource group
-        rg = (source_vm[0] if source_vm else dest_vm[0])
+        rg_provided = (source_vm[0] if source_vm else dest_vm[0])
         vm_name = (source_vm[1] if source_vm else dest_vm[1])
 
-        if not rg:
-            # Get RG of the VM
-            client = _compute_client_factory(cmd.cli_ctx)
-            vms = client.virtual_machines.list_all()
-            vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
-            if not vm:
-                raise ResourceNotFoundError("VM '{}' not found.".format(vm_name))
-            # parse RG from ID
-            rg = vm.id.split('/')[4]
+        _, rg = _get_vm_and_rg(cmd, vm_name, rg_provided)
 
         from azure.cli.command_modules.storage._client_factory import cf_sa
         sa_client = cf_sa(cmd.cli_ctx, None)
@@ -6666,13 +6674,8 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
     try:
         if dest_vm:
             # UPLOAD: Local -> VM
-            rg, vm_name, vm_path = dest_vm
-            if not rg:
-                # find VM RG
-                client = _compute_client_factory(cmd.cli_ctx)
-                vms = client.virtual_machines.list_all()
-                vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
-                rg = vm.id.split('/')[4]
+            rg_provided, vm_name, vm_path = dest_vm
+            vm_obj, rg = _get_vm_and_rg(cmd, vm_name, rg_provided)
 
             logger.info("Uploading local file to bridge storage...")
             upload_blob(cmd, blob_client, file_path=source)
@@ -6683,7 +6686,6 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
 
             # VM run-command to download
             # Check OS type
-            vm_obj = _compute_client_factory(cmd.cli_ctx).virtual_machines.get(rg, vm_name)
             is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
 
             if is_linux:
@@ -6710,13 +6712,8 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
 
         else:
             # DOWNLOAD: VM -> Local
-            rg, vm_name, vm_path = source_vm
-            if not rg:
-                # find VM RG
-                client = _compute_client_factory(cmd.cli_ctx)
-                vms = client.virtual_machines.list_all()
-                vm = next((v for v in vms if v.name.lower() == vm_name.lower()), None)
-                rg = vm.id.split('/')[4]
+            rg_provided, vm_name, vm_path = source_vm
+            vm_obj, rg = _get_vm_and_rg(cmd, vm_name, rg_provided)
 
             # Get SAS with WRITE permission (2 hours expiry)
             t_sas = cmd.get_models('_shared_access_signature#BlobSharedAccessSignature',
@@ -6729,7 +6726,6 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
                                           expiry=expiry, protocol='https')
             blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
 
-            vm_obj = _compute_client_factory(cmd.cli_ctx).virtual_machines.get(rg, vm_name)
             is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
 
             if is_linux:
@@ -6739,10 +6735,10 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
                 escaped_vm_path = vm_path.replace("'", "''")
                 escaped_blob_url = blob_url.replace("'", "''")
                 # Cross-version compatible Get-Content as suggested by Copilot
-                script = ("$body = if ($PSVersionTable.PSVersion.Major -ge 6) {{ Get-Content -Path '{}' -AsByteStream }} "
-                          "else {{ Get-Content -Path '{}' -Encoding Byte }}; "
-                          "Invoke-RestMethod -Uri '{}' -Method Put -Headers @{{'x-ms-blob-type'='BlockBlob'}} -Body $body").format(
-                              escaped_vm_path, escaped_vm_path, escaped_blob_url)
+                script = ("$body = if ($PSVersionTable.PSVersion.Major -ge 6) {{ Get-Content -Path '{path}' -AsByteStream }} "
+                          "else {{ Get-Content -Path '{path}' -Encoding Byte }}; "
+                          "Invoke-RestMethod -Uri '{url}' -Method Put -Headers @{{'x-ms-blob-type'='BlockBlob'}} -Body $body").format(
+                              path=escaped_vm_path, url=escaped_blob_url)
                 command_id = 'RunPowerShellScript'
 
             logger.info("Executing upload script in VM...")

From 8d3bd367b9375920259f024402efe38ab7af4331 Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Fri, 6 Feb 2026 20:24:19 +0530
Subject: [PATCH 07/12] fix(vm): address pylint style issues in custom.py

---
 .../azure/cli/command_modules/vm/custom.py    | 39 +++++++------------
 1 file changed, 14 insertions(+), 25 deletions(-)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index 3823bb51cce..4beae7bce93 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -116,7 +116,6 @@ def _construct_identity_info(identity_scope, identity_role, implicit_identity, e
 
 # for injecting test seams to produce predicatable role assignment id for playback
 def _gen_guid():
-    import uuid
     return uuid.uuid4()
 
 
@@ -3432,8 +3431,7 @@ def attach_unmanaged_data_disk(cmd, resource_group_name, vm_name, new=False, vhd
 
     vm = get_vm_to_update(cmd, resource_group_name, vm_name)
     if disk_name is None:
-        import datetime
-        disk_name = vm_name + '-' + datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
+        disk_name = vm_name + '-' + datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
     # pylint: disable=no-member
     if vhd_uri is None:
         if not hasattr(vm.storage_profile.os_disk, 'vhd') or not vm.storage_profile.os_disk.vhd:
@@ -6559,37 +6557,28 @@ def _is_windows_absolute_path(path):
 
 
 def _parse_vm_file_path(path):
-    # If there is no colon, this cannot be a VM path.
-    if ':' not in path:
+    if ':' not in path or _is_windows_absolute_path(path):
         return None
 
-    # Do not treat Windows absolute paths like 'C:\\path\\to\\file' as VM paths.
-    if _is_windows_absolute_path(path):
-        return None
-
-    # VM path format is [resource-group-name:]vm-name:path-on-vm
-    # Only the first two colons separate components; any remaining colons belong to the path.
     first_colon = path.find(':')
+    # second_colon - find the next colon after the first one
     second_colon = path.find(':', first_colon + 1)
 
-    if first_colon == -1:
-        return None
-
     if second_colon == -1:
         # vm-name:path
         vm_name = path[:first_colon]
         vm_path = path[first_colon + 1:]
-        if not vm_name or not vm_path:
-            return None
-        return None, vm_name, vm_path
-
-    # rg:vm-name:path...
-    rg_name = path[:first_colon]
-    vm_name = path[first_colon + 1:second_colon]
-    vm_path = path[second_colon + 1:]
-    if not rg_name or not vm_name or not vm_path:
-        return None
-    return rg_name, vm_name, vm_path
+        if vm_name and vm_path:
+            return None, vm_name, vm_path
+    else:
+        # rg:vm-name:path...
+        rg_name = path[:first_colon]
+        vm_name = path[first_colon + 1:second_colon]
+        vm_path = path[second_colon + 1:]
+        if rg_name and vm_name and vm_path:
+            return rg_name, vm_name, vm_path
+
+    return None
 
 
 def _get_vm_and_rg(cmd, vm_name, rg=None):

From 71ab8fa91342c9cd7686993324c3cf78b4fd1b9b Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Mon, 9 Feb 2026 12:37:09 +0530
Subject: [PATCH 08/12] refactor(vm): move inline imports to top and fix unit
 tests for 'az vm cp'

---
 .../azure/cli/command_modules/vm/custom.py          | 13 +++++++++----
 .../vm/tests/latest/test_vm_cp_unit.py              |  3 ++-
 2 files changed, 11 insertions(+), 5 deletions(-)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index 4beae7bce93..e136590d852 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -53,7 +53,9 @@
 
 from .aaz.latest.vm.disk import AttachDetachDataDisk
 from .aaz.latest.vm import Update as UpdateVM
+from .aaz.latest.vm.run_command import Invoke
 
+from azure.cli.command_modules.storage._client_factory import cf_sa, cf_sa_for_keys, cf_blob_service
 from .generated.custom import *  # noqa: F403, pylint: disable=unused-wildcard-import,wildcard-import
 try:
     from .manual.custom import *   # noqa: F403, pylint: disable=unused-wildcard-import,wildcard-import
@@ -6571,6 +6573,13 @@ def _parse_vm_file_path(path):
         if vm_name and vm_path:
             return None, vm_name, vm_path
     else:
+        # Check if it's vm:C:\path (second colon is a drive letter)
+        if second_colon - first_colon == 2 and path[first_colon + 1].isalpha():
+            vm_name = path[:first_colon]
+            vm_path = path[first_colon + 1:]
+            if vm_name and vm_path:
+                return None, vm_name, vm_path
+
         # rg:vm-name:path...
         rg_name = path[:first_colon]
         vm_name = path[first_colon + 1:second_colon]
@@ -6598,7 +6607,6 @@ def _get_vm_and_rg(cmd, vm_name, rg=None):
 
 
 def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp'):
-    from .aaz.latest.vm.run_command import Invoke
 
     source_vm = _parse_vm_file_path(source)
     dest_vm = _parse_vm_file_path(destination)
@@ -6616,7 +6624,6 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
 
         _, rg = _get_vm_and_rg(cmd, vm_name, rg_provided)
 
-        from azure.cli.command_modules.storage._client_factory import cf_sa
         sa_client = cf_sa(cmd.cli_ctx, None)
         accounts = list(sa_client.list())
         # Filter by RG if possible
@@ -6629,7 +6636,6 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
             raise RequiredArgumentMissingError("No storage account found in the subscription. Please provide one with --storage-account.")
 
     # Get account key
-    from azure.cli.command_modules.storage._client_factory import cf_sa_for_keys
     sa_keys_client = cf_sa_for_keys(cmd.cli_ctx, None)
     # Check if storage_account is name or ID
     if '/' in storage_account:
@@ -6649,7 +6655,6 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
     account_key = keys[0].value
 
     # Ensure container exists
-    from azure.cli.command_modules.storage._client_factory import cf_blob_service
     blob_service_client = cf_blob_service(cmd.cli_ctx, {'account_name': sa_name, 'account_key': account_key})
     container_client = blob_service_client.get_container_client(container_name)
     try:
diff --git a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
index b61689db6bc..03588de8b66 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
@@ -41,6 +41,7 @@ def test_vm_cp_upload_basic(self, mock_upload, mock_sas, mock_blob_factory, mock
         mock_compute_factory.return_value = mock_compute
         
         vm_obj = mock.MagicMock()
+        vm_obj.name = "myvm"
         vm_obj.storage_profile.os_disk.os_type.lower.return_value = 'linux'
         vm_obj.id = "/subscriptions/sub/resourceGroups/myrg/providers/Microsoft.Compute/virtualMachines/myvm"
         mock_compute.virtual_machines.get.return_value = vm_obj
@@ -72,7 +73,7 @@ def test_vm_cp_upload_basic(self, mock_upload, mock_sas, mock_blob_factory, mock
         # Execute
         with mock.patch('azure.cli.command_modules.vm.custom.Invoke') as mock_invoke:
             mock_invoke.return_value.return_value = {'value': [{'message': 'success'}]}
-            vm_cp(cmd, source="local.txt", destination="myvm:/tmp/remote.txt")
+            vm_cp(cmd, source="local.txt", destination="myrg:myvm:/tmp/remote.txt")
             
             # Verify
             mock_upload.assert_called_once()

From b6456c0124a6136bd1a46a9fb38c70b1c15d189e Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Mon, 9 Feb 2026 12:40:40 +0530
Subject: [PATCH 09/12] refactor(vm): reduce code duplication in vm_cp and add
 download unit test

---
 .../azure/cli/command_modules/vm/custom.py    | 83 ++++++++-----------
 .../vm/tests/latest/test_vm_cp_unit.py        | 53 ++++++++++++
 2 files changed, 88 insertions(+), 48 deletions(-)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index e136590d852..dddcf0ded60 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -6665,50 +6665,22 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
     blob_name = str(uuid.uuid4())
     blob_client = container_client.get_blob_client(blob_name)
 
+    # Common VM info
+    vm_info = dest_vm if dest_vm else source_vm
+    rg_provided, vm_name, vm_path = vm_info
+    vm_obj, rg = _get_vm_and_rg(cmd, vm_name, rg_provided)
+    is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
+
     try:
         if dest_vm:
             # UPLOAD: Local -> VM
-            rg_provided, vm_name, vm_path = dest_vm
-            vm_obj, rg = _get_vm_and_rg(cmd, vm_name, rg_provided)
-
             logger.info("Uploading local file to bridge storage...")
             upload_blob(cmd, blob_client, file_path=source)
 
-            # Get SAS for VM to download (2 hours expiry)
+            # Get SAS with READ permission (2 hours expiry)
             sas_token = create_short_lived_blob_sas_v2(cmd, sa_name, container_name, blob_name, account_key=account_key)
-            blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
-
-            # VM run-command to download
-            # Check OS type
-            is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
-
-            if is_linux:
-                script = "curl -L -f -s -S -o {} {}".format(shlex.quote(vm_path), shlex.quote(blob_url))
-                command_id = 'RunShellScript'
-            else:
-                # Escape single quotes for PowerShell
-                escaped_vm_path = vm_path.replace("'", "''")
-                escaped_blob_url = blob_url.replace("'", "''")
-                script = "Invoke-WebRequest -Uri '{}' -OutFile '{}'".format(escaped_blob_url, escaped_vm_path)
-                command_id = 'RunPowerShellScript'
-
-            logger.info("Executing download script in VM...")
-            result = Invoke(cli_ctx=cmd.cli_ctx)(command_args={
-                'resource_group': rg,
-                'vm_name': vm_name,
-                'command_id': command_id,
-                'script': [script]
-            })
-            if result.get('value') and result['value'][0].get('message'):
-                message = result['value'][0]['message']
-                if 'failed' in message.lower() or 'error' in message.lower():
-                    raise CLIError("VM execution failed: {}".format(message))
-
         else:
             # DOWNLOAD: VM -> Local
-            rg_provided, vm_name, vm_path = source_vm
-            vm_obj, rg = _get_vm_and_rg(cmd, vm_name, rg_provided)
-
             # Get SAS with WRITE permission (2 hours expiry)
             t_sas = cmd.get_models('_shared_access_signature#BlobSharedAccessSignature',
                                    resource_type=ResourceType.DATA_STORAGE_BLOB)
@@ -6718,10 +6690,22 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
             sas_token = sas.generate_blob(container_name, blob_name,
                                           permission=t_blob_permissions(write=True),
                                           expiry=expiry, protocol='https')
-            blob_url = "https://{}.blob.{}/{}/{}?{}".format(sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
 
-            is_linux = vm_obj.storage_profile.os_disk.os_type.lower() == 'linux'
+        blob_url = "https://{}.blob.{}/{}/{}?{}".format(
+            sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
 
+        if dest_vm:
+            # Script for VM to download from blob
+            if is_linux:
+                script = "curl -L -f -s -S -o {} {}".format(shlex.quote(vm_path), shlex.quote(blob_url))
+                command_id = 'RunShellScript'
+            else:
+                escaped_vm_path = vm_path.replace("'", "''")
+                escaped_blob_url = blob_url.replace("'", "''")
+                script = "Invoke-WebRequest -Uri '{}' -OutFile '{}'".format(escaped_blob_url, escaped_vm_path)
+                command_id = 'RunPowerShellScript'
+        else:
+            # Script for VM to upload to blob
             if is_linux:
                 script = "curl -X PUT -T {} -H 'x-ms-blob-type: BlockBlob' {}".format(shlex.quote(vm_path), shlex.quote(blob_url))
                 command_id = 'RunShellScript'
@@ -6735,21 +6719,24 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
                               path=escaped_vm_path, url=escaped_blob_url)
                 command_id = 'RunPowerShellScript'
 
-            logger.info("Executing upload script in VM...")
-            result = Invoke(cli_ctx=cmd.cli_ctx)(command_args={
-                'resource_group': rg,
-                'vm_name': vm_name,
-                'command_id': command_id,
-                'script': [script]
-            })
-            if result.get('value') and result['value'][0].get('message'):
-                message = result['value'][0]['message']
-                if 'failed' in message.lower() or 'error' in message.lower():
-                    raise CLIError("VM execution failed: {}".format(message))
+        logger.info("Executing transfer script in VM...")
+        result = Invoke(cli_ctx=cmd.cli_ctx)(command_args={
+            'resource_group': rg,
+            'vm_name': vm_name,
+            'command_id': command_id,
+            'script': [script]
+        })
 
+        if result.get('value') and result['value'][0].get('message'):
+            message = result['value'][0]['message']
+            if 'failed' in message.lower() or 'error' in message.lower():
+                raise CLIError("VM execution failed: {}".format(message))
+
+        if not dest_vm:
             logger.info("Downloading from bridge storage to local...")
             download_blob(blob_client, file_path=destination)
 
+
     finally:
         # Cleanup bridge storage
         try:
diff --git a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
index 03588de8b66..d597e779fcd 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
@@ -80,5 +80,58 @@ def test_vm_cp_upload_basic(self, mock_upload, mock_sas, mock_blob_factory, mock
             mock_compute.virtual_machines.get.assert_called()
             mock_blob.delete_blob.assert_called_once()
 
+    @mock.patch('azure.cli.command_modules.vm.custom._compute_client_factory')
+    @mock.patch('azure.cli.command_modules.vm.custom.cf_sa')
+    @mock.patch('azure.cli.command_modules.vm.custom.cf_sa_for_keys')
+    @mock.patch('azure.cli.command_modules.vm.custom.cf_blob_service')
+    @mock.patch('azure.cli.command_modules.vm.custom.download_blob')
+    def test_vm_cp_download_basic(self, mock_download, mock_blob_factory, mock_keys_factory, mock_sa_factory, mock_compute_factory):
+        cmd = mock.MagicMock()
+        cmd.cli_ctx.cloud.suffixes.storage_endpoint = 'core.windows.net'
+        
+        # Mock compute client
+        mock_compute = mock.MagicMock()
+        mock_compute_factory.return_value = mock_compute
+        
+        vm_obj = mock.MagicMock()
+        vm_obj.name = "myvm"
+        vm_obj.storage_profile.os_disk.os_type.lower.return_value = 'linux'
+        vm_obj.id = "/subscriptions/sub/resourceGroups/myrg/providers/Microsoft.Compute/virtualMachines/myvm"
+        mock_compute.virtual_machines.get.return_value = vm_obj
+        
+        # Mock storage clients
+        mock_sa = mock.MagicMock()
+        mock_sa_factory.return_value = mock_sa
+        
+        sa = mock.MagicMock()
+        sa.name = 'mystorage'
+        sa.id = "/subscriptions/sub/resourceGroups/myrg/providers/Microsoft.Storage/storageAccounts/mystorage"
+        mock_sa.list.return_value = [sa]
+        
+        mock_keys = mock.MagicMock()
+        mock_keys_factory.return_value = mock_keys
+        key = mock.MagicMock()
+        key.value = 'key1'
+        mock_keys.list_keys.return_value.keys = [key]
+        
+        # Mock blob service
+        mock_blob_service = mock.MagicMock()
+        mock_blob_factory.return_value = mock_blob_service
+        mock_container = mock.MagicMock()
+        mock_blob_service.get_container_client.return_value = mock_container
+        mock_blob = mock.MagicMock()
+        mock_container.get_blob_client.return_value = mock_blob
+        
+        # Execute
+        with mock.patch('azure.cli.command_modules.vm.custom.Invoke') as mock_invoke:
+            mock_invoke.return_value.return_value = {'value': [{'message': 'success'}]}
+            vm_cp(cmd, source="myrg:myvm:/tmp/remote.txt", destination="local.txt")
+            
+            # Verify
+            mock_invoke.assert_called_once()
+            mock_download.assert_called_once()
+            mock_blob.delete_blob.assert_called_once()
+
 if __name__ == '__main__':
     unittest.main()
+

From 85b4e383b991077c2b6006ce6ce48d50ad8534d9 Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Mon, 9 Feb 2026 13:01:02 +0530
Subject: [PATCH 10/12] fix(vm): fix CI loading issue by lazy-importing
 'storage' and improve URL robustness

---
 .../azure/cli/command_modules/vm/custom.py     | 15 ++++++++++-----
 .../vm/tests/latest/test_vm_cp_unit.py         | 18 +++++++++---------
 2 files changed, 19 insertions(+), 14 deletions(-)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index dddcf0ded60..07932f2445f 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -44,9 +44,6 @@
 from ._vm_utils import read_content_if_is_file, import_aaz_by_profile, IdentityType
 from ._vm_diagnostics_templates import get_default_diag_config
 
-from azure.cli.command_modules.storage.operations.blob import upload_blob, download_blob
-from azure.cli.command_modules.storage.util import create_short_lived_blob_sas_v2
-
 from ._actions import (load_images_from_aliases_doc, load_extension_images_thru_services,
                        load_images_thru_services, _get_latest_image_version, _get_latest_image_version_by_aaz)
 from ._client_factory import (_compute_client_factory, cf_vm_image_term)
@@ -55,7 +52,6 @@
 from .aaz.latest.vm import Update as UpdateVM
 from .aaz.latest.vm.run_command import Invoke
 
-from azure.cli.command_modules.storage._client_factory import cf_sa, cf_sa_for_keys, cf_blob_service
 from .generated.custom import *  # noqa: F403, pylint: disable=unused-wildcard-import,wildcard-import
 try:
     from .manual.custom import *   # noqa: F403, pylint: disable=unused-wildcard-import,wildcard-import
@@ -6607,6 +6603,9 @@ def _get_vm_and_rg(cmd, vm_name, rg=None):
 
 
 def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp'):
+    from azure.cli.command_modules.storage.operations.blob import upload_blob, download_blob
+    from azure.cli.command_modules.storage.util import create_short_lived_blob_sas_v2
+    from azure.cli.command_modules.storage._client_factory import cf_sa, cf_sa_for_keys, cf_blob_service
 
     source_vm = _parse_vm_file_path(source)
     dest_vm = _parse_vm_file_path(destination)
@@ -6691,8 +6690,14 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
                                           permission=t_blob_permissions(write=True),
                                           expiry=expiry, protocol='https')
 
+        storage_suffix = getattr(cmd.cli_ctx.cloud.suffixes, 'storage_endpoint', None)
+        if not storage_suffix:
+            raise CLIError("The storage endpoint suffix for the current cloud is not configured.")
+        # Normalize to avoid leading dots or slashes that would break the URL host
+        storage_suffix = str(storage_suffix).lstrip("./").strip()
+
         blob_url = "https://{}.blob.{}/{}/{}?{}".format(
-            sa_name, cmd.cli_ctx.cloud.suffixes.storage_endpoint, container_name, blob_name, sas_token)
+            sa_name, storage_suffix, container_name, blob_name, sas_token)
 
         if dest_vm:
             # Script for VM to download from blob
diff --git a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
index d597e779fcd..af53caf14d7 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
@@ -27,11 +27,11 @@ def test_parse_vm_file_path(self):
         self.assertEqual(_parse_vm_file_path("rg:vm:C:\\remote\\path"), ("rg", "vm", "C:\\remote\\path"))
 
     @mock.patch('azure.cli.command_modules.vm.custom._compute_client_factory')
-    @mock.patch('azure.cli.command_modules.vm.custom.cf_sa')
-    @mock.patch('azure.cli.command_modules.vm.custom.cf_sa_for_keys')
-    @mock.patch('azure.cli.command_modules.vm.custom.cf_blob_service')
-    @mock.patch('azure.cli.command_modules.vm.custom.create_short_lived_blob_sas_v2')
-    @mock.patch('azure.cli.command_modules.vm.custom.upload_blob')
+    @mock.patch('azure.cli.command_modules.storage._client_factory.cf_sa')
+    @mock.patch('azure.cli.command_modules.storage._client_factory.cf_sa_for_keys')
+    @mock.patch('azure.cli.command_modules.storage._client_factory.cf_blob_service')
+    @mock.patch('azure.cli.command_modules.storage.util.create_short_lived_blob_sas_v2')
+    @mock.patch('azure.cli.command_modules.storage.operations.blob.upload_blob')
     def test_vm_cp_upload_basic(self, mock_upload, mock_sas, mock_blob_factory, mock_keys_factory, mock_sa_factory, mock_compute_factory):
         cmd = mock.MagicMock()
         cmd.cli_ctx.cloud.suffixes.storage_endpoint = 'core.windows.net'
@@ -81,10 +81,10 @@ def test_vm_cp_upload_basic(self, mock_upload, mock_sas, mock_blob_factory, mock
             mock_blob.delete_blob.assert_called_once()
 
     @mock.patch('azure.cli.command_modules.vm.custom._compute_client_factory')
-    @mock.patch('azure.cli.command_modules.vm.custom.cf_sa')
-    @mock.patch('azure.cli.command_modules.vm.custom.cf_sa_for_keys')
-    @mock.patch('azure.cli.command_modules.vm.custom.cf_blob_service')
-    @mock.patch('azure.cli.command_modules.vm.custom.download_blob')
+    @mock.patch('azure.cli.command_modules.storage._client_factory.cf_sa')
+    @mock.patch('azure.cli.command_modules.storage._client_factory.cf_sa_for_keys')
+    @mock.patch('azure.cli.command_modules.storage._client_factory.cf_blob_service')
+    @mock.patch('azure.cli.command_modules.storage.operations.blob.download_blob')
     def test_vm_cp_download_basic(self, mock_download, mock_blob_factory, mock_keys_factory, mock_sa_factory, mock_compute_factory):
         cmd = mock.MagicMock()
         cmd.cli_ctx.cloud.suffixes.storage_endpoint = 'core.windows.net'

From 646a284f8184ec0b81b7f5954ad5343c3e679ae9 Mon Sep 17 00:00:00 2001
From: ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Mon, 9 Feb 2026 14:39:27 +0530
Subject: [PATCH 11/12] fix(vm): resolve pylint W0621 and flake8 E303 style
 issues

---
 src/azure-cli/azure/cli/command_modules/vm/custom.py | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index 07932f2445f..d79d0a93fe9 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -50,7 +50,6 @@
 
 from .aaz.latest.vm.disk import AttachDetachDataDisk
 from .aaz.latest.vm import Update as UpdateVM
-from .aaz.latest.vm.run_command import Invoke
 
 from .generated.custom import *  # noqa: F403, pylint: disable=unused-wildcard-import,wildcard-import
 try:
@@ -6606,6 +6605,7 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
     from azure.cli.command_modules.storage.operations.blob import upload_blob, download_blob
     from azure.cli.command_modules.storage.util import create_short_lived_blob_sas_v2
     from azure.cli.command_modules.storage._client_factory import cf_sa, cf_sa_for_keys, cf_blob_service
+    from .aaz.latest.vm.run_command import Invoke
 
     source_vm = _parse_vm_file_path(source)
     dest_vm = _parse_vm_file_path(destination)
@@ -6736,12 +6736,9 @@ def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp
             message = result['value'][0]['message']
             if 'failed' in message.lower() or 'error' in message.lower():
                 raise CLIError("VM execution failed: {}".format(message))
-
         if not dest_vm:
             logger.info("Downloading from bridge storage to local...")
             download_blob(blob_client, file_path=destination)
-
-
     finally:
         # Cleanup bridge storage
         try:

From c28574a06261fe1df38bbd7ba1f28e8a042f3add Mon Sep 17 00:00:00 2001
From: Ashutosh0x <ashutoshkumarsingh0x@gmail.com>
Date: Thu, 5 Mar 2026 17:07:31 +0530
Subject: [PATCH 12/12] fix(vm): lazy-import shlex/ResourceExistsError and fix
 Invoke mock target

---
 src/azure-cli/azure/cli/command_modules/vm/custom.py         | 5 +++--
 .../cli/command_modules/vm/tests/latest/test_vm_cp_unit.py   | 4 ++--
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/src/azure-cli/azure/cli/command_modules/vm/custom.py b/src/azure-cli/azure/cli/command_modules/vm/custom.py
index d79d0a93fe9..8b3974827fa 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/custom.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/custom.py
@@ -15,7 +15,6 @@
 import json
 import os
 import uuid
-import shlex
 from datetime import datetime, timedelta
 
 import requests
@@ -25,7 +24,7 @@
 
 from knack.log import get_logger
 from knack.util import CLIError
-from azure.core.exceptions import ResourceExistsError
+
 from azure.cli.core.azclierror import (
     ResourceNotFoundError,
     ValidationError,
@@ -6602,6 +6601,8 @@ def _get_vm_and_rg(cmd, vm_name, rg=None):
 
 
 def vm_cp(cmd, source, destination, storage_account=None, container_name='azvmcp'):
+    import shlex
+    from azure.core.exceptions import ResourceExistsError
     from azure.cli.command_modules.storage.operations.blob import upload_blob, download_blob
     from azure.cli.command_modules.storage.util import create_short_lived_blob_sas_v2
     from azure.cli.command_modules.storage._client_factory import cf_sa, cf_sa_for_keys, cf_blob_service
diff --git a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
index af53caf14d7..4b431dd5996 100644
--- a/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
+++ b/src/azure-cli/azure/cli/command_modules/vm/tests/latest/test_vm_cp_unit.py
@@ -71,7 +71,7 @@ def test_vm_cp_upload_basic(self, mock_upload, mock_sas, mock_blob_factory, mock
         mock_container.get_blob_client.return_value = mock_blob
         
         # Execute
-        with mock.patch('azure.cli.command_modules.vm.custom.Invoke') as mock_invoke:
+        with mock.patch('azure.cli.command_modules.vm.aaz.latest.vm.run_command.Invoke') as mock_invoke:
             mock_invoke.return_value.return_value = {'value': [{'message': 'success'}]}
             vm_cp(cmd, source="local.txt", destination="myrg:myvm:/tmp/remote.txt")
             
@@ -123,7 +123,7 @@ def test_vm_cp_download_basic(self, mock_download, mock_blob_factory, mock_keys_
         mock_container.get_blob_client.return_value = mock_blob
         
         # Execute
-        with mock.patch('azure.cli.command_modules.vm.custom.Invoke') as mock_invoke:
+        with mock.patch('azure.cli.command_modules.vm.aaz.latest.vm.run_command.Invoke') as mock_invoke:
             mock_invoke.return_value.return_value = {'value': [{'message': 'success'}]}
             vm_cp(cmd, source="myrg:myvm:/tmp/remote.txt", destination="local.txt")