[KeyVault] az keyvault secret copy: Add command to copy secrets between vaults

src/azure-cli/azure/cli/command_modules/keyvault/_help.py

@@ -906,6 +906,20 @@
     the secret will be downloaded. This operation requires the secrets/backup permission.
 """
 
+helps['keyvault secret copy'] = """
+type: command
+short-summary: Copy a secret from one Key Vault to another.
+long-summary: Copies the latest version of a secret from a source Key Vault to a destination Key Vault.
+    This operation copies the secret value and its metadata (tags, content-type, attributes).
+examples:
+  - name: Copy a specific secret from one vault to another.
+    text: az keyvault secret copy --source-vault SourceVault --destination-vault DestVault --name MySecret
+  - name: Copy all secrets from one vault to another.
+    text: az keyvault secret copy --source-vault SourceVault --destination-vault DestVault --all
+  - name: Copy a secret and overwrite if it already exists in the destination.
+    text: az keyvault secret copy --source-vault SourceVault --destination-vault DestVault --name MySecret --overwrite
+"""
+
 helps['keyvault secret restore'] = """
 type: command
 short-summary: Restores a backed up secret to a vault.

src/azure-cli/azure/cli/command_modules/keyvault/_params.py

@@ -568,6 +568,20 @@ class CLISecurityDomainOperation(str, Enum):
     with self.argument_context('keyvault secret restore') as c:
         c.extra('vault_base_url', vault_name_type, required=True, arg_group='Id',
                 type=get_vault_base_url_type(self.cli_ctx), id_part=None)
+
+    with self.argument_context('keyvault secret copy') as c:
+        c.extra('vault_base_url', vault_name_type, type=get_vault_base_url_type(self.cli_ctx),
+                options_list=['--source-vault'], help='Name of the source Key Vault.', required=True)
+        c.extra('destination_vault', vault_name_type, type=get_vault_base_url_type(self.cli_ctx),
+                options_list=['--destination-vault'], help='Name of the destination Key Vault.', required=True)
+        c.argument('name', options_list=['--name', '-n'],
+                   help='Name of the secret to copy. Mutually exclusive with --all. If neither --name nor --all is '
+                        'specified, all secrets will be copied.',
+                   required=False)
+        c.extra('all_secrets', arg_type=get_three_state_flag(), options_list=['--all'],
+                help='Copy all secrets from the source vault. Mutually exclusive with --name. If neither --name nor '
+                     '--all is specified, all secrets will be copied.')
+        c.extra('overwrite', arg_type=get_three_state_flag(), help='Overwrite secrets in the destination vault if they already exist.')
     # endregion
 
     # region keyvault security-domain

src/azure-cli/azure/cli/command_modules/keyvault/commands.py

@@ -204,6 +204,7 @@ def load_command_table(self, _):
         g.keyvault_custom('download', 'download_secret')
         g.keyvault_custom('backup', 'backup_secret')
         g.keyvault_custom('restore', 'restore_secret', transform=transform_secret_set_attributes)
+        g.keyvault_custom('copy', 'copy_secret')
 
     # certificate track2
     with self.command_group('keyvault certificate', data_certificate_entity.command_type) as g:

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

@@ -2517,3 +2517,120 @@ def set_attributes_certificate(client, certificate_name, version=None, policy=No
     if kwargs.get('enabled') is not None or kwargs.get('tags') is not None:
         return client.update_certificate_properties(certificate_name=certificate_name, version=version, **kwargs)
     return client.get_certificate(certificate_name=certificate_name)
+
+
+def _copy_single_secret(source_client, dest_client, secret_name, overwrite, is_single_mode):
+    from azure.core.exceptions import ResourceNotFoundError, HttpResponseError
+
+    try:
+        # Check destination
+        if not overwrite:
+            try:
+                dest_client.get_secret(secret_name)
+                logger.warning("Secret '%s' already exists in destination. Skipping.", secret_name)
+                return None  # Skipped
+            except ResourceNotFoundError:
+                pass
+            except HttpResponseError as e:
+                logger.warning("Error checking secret '%s' in destination: %s", secret_name, str(e))
+                return False  # Failed
+
+        # Copy
+        logger.info("Copying secret: %s", secret_name)
+        s = source_client.get_secret(secret_name)
+
+        try:
+            new_secret = dest_client.set_secret(
+                s.name,
+                s.value,
+                content_type=s.properties.content_type,
+                tags=s.properties.tags,
+                enabled=s.properties.enabled,
+                not_before=s.properties.not_before,
+                expires_on=s.properties.expires_on
+            )
+        except HttpResponseError as e:
+            if is_single_mode:
+                raise e
+
+            logger.error("Failed to copy secret '%s': %s", secret_name, str(e))
+            return False
+
+        logger.info("Successfully copied secret: %s", secret_name)
+        return {'name': new_secret.name, 'id': new_secret.id}
+
+    except ResourceNotFoundError as e:
+        if is_single_mode:
+            raise e
+        logger.error("Secret '%s' not found in source vault.", secret_name)
+        return False
+    except HttpResponseError as e:
+        if is_single_mode:
+            raise e
+        logger.error("Failed to copy secret '%s': %s", secret_name, str(e))
+        return False
+
+
+def copy_secret(cmd, client, destination_vault, name=None, all_secrets=None, overwrite=False):
+    from azure.core.exceptions import ResourceNotFoundError, HttpResponseError
+    from azure.cli.command_modules.keyvault._client_factory import data_plane_azure_keyvault_secret_client
+
+    # If neither a specific secret name nor --all is provided, default to copying all secrets.
+    if not name and not all_secrets:
+        all_secrets = True
+
+    # A specific secret name and --all are mutually exclusive.
+    if name and all_secrets:
+        raise MutuallyExclusiveArgumentError("Specify either a secret name or --all, but not both.")
+    # Validation
+    if client.vault_url.rstrip('/') == destination_vault.rstrip('/'):
+        raise CLIError("Source and destination Key Vaults cannot be the same.")
+
+    command_args = {'vault_base_url': destination_vault}
+    dest_client = data_plane_azure_keyvault_secret_client(cmd.cli_ctx, command_args)
+
+    # Fail fast if source or destination vault is not accessible or does not exist
+    for c, vault_url in [(client, client.vault_url), (dest_client, destination_vault)]:
+        try:
+            # Perform a lightweight call to validate vault accessibility.
+            # A 404 for a dummy secret name means the vault is reachable but the secret does not exist.
+            c.get_secret("azure-cli-validation-dummy")
+        except ResourceNotFoundError:
+            # Vault is accessible but the dummy secret does not exist, which is expected.
+            pass
+        except HttpResponseError as e:
+            if e.status_code == 404:
+                # Vault is accessible but the dummy secret does not exist, which is expected.
+                pass
+            else:
+                raise CLIError(f"Failed to access Key Vault '{vault_url}': {str(e)}")
+
+    secrets_to_copy = []
+    if name:
+        secrets_to_copy.append(name)
+    else:
+        logger.info("Copying all secrets from source...")
+        try:
+            source_secrets = client.list_properties_of_secrets()
+            for s in source_secrets:
+                if s.managed:
+                    logger.warning("Skipping managed secret: %s", s.name)
+                    continue
+                secrets_to_copy.append(s.name)
+        except HttpResponseError as e:
+            raise CLIError(f"Failed to list secrets from source: {str(e)}")
+
+    copied_secrets = []
+    failed_secrets = []
+    for secret_name in secrets_to_copy:
+        result = _copy_single_secret(client, dest_client, secret_name, overwrite, bool(name))
+        if result:
+            copied_secrets.append(result)
+        elif result is False:
+            failed_secrets.append(secret_name)
+
+    if failed_secrets:
+        logger.warning("Operation completed with failures. %s secrets failed to copy: %s",
+                       len(failed_secrets), ', '.join(failed_secrets))
+
+    return copied_secrets