[Key Vault] Add External Key Manager (EKM) support for Managed HSM (Preview)

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

@@ -244,6 +244,26 @@ def data_plane_azure_keyvault_security_domain_client(cli_ctx, command_args):
                                 verify_challenge_resource=False, **client_kwargs)
 
 
+def data_plane_azure_keyvault_ekm_client(cli_ctx, command_args):
+    from azure.keyvault.administration import KeyVaultEkmClient
+
+    # Reuse the existing login + URL resolution behavior.
+    vault_url, credential, _ = _prepare_data_plane_azure_keyvault_client(
+        cli_ctx, command_args, ResourceType.DATA_KEYVAULT_ADMINISTRATION_SETTING)
+
+    command_args.pop('hsm_name', None)
+    command_args.pop('vault_base_url', None)
+    command_args.pop('identifier', None)
+
+    client_kwargs = prepare_client_kwargs_track2(cli_ctx)
+    client_kwargs.pop('http_logging_policy')
+    return KeyVaultEkmClient(
+        vault_url=vault_url,
+        credential=credential,
+        verify_challenge_resource=False,
+        **client_kwargs)
+
+
 def _prepare_data_plane_azure_keyvault_client(cli_ctx, command_args, resource_type):
     version = str(get_api_version(cli_ctx, resource_type))
     profile = Profile(cli_ctx=cli_ctx)

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

@@ -975,6 +975,46 @@
         az keyvault wait-hsm --hsm-name MyHSM --created
 """
 
+helps['keyvault ekm-connection'] = """
+type: group
+short-summary: Manage External Key Manager (EKM) connection for a Managed HSM.
+"""
+
+helps['keyvault ekm-connection create'] = """
+type: command
+short-summary: Create the EKM connection.
+"""
+
+helps['keyvault ekm-connection update'] = """
+type: command
+short-summary: Update the EKM connection.
+"""
+
+helps['keyvault ekm-connection show'] = """
+type: command
+short-summary: Show the EKM connection.
+"""
+
+helps['keyvault ekm-connection check'] = """
+type: command
+short-summary: Check connectivity and authentication with the EKM proxy.
+"""
+
+helps['keyvault ekm-connection delete'] = """
+type: command
+short-summary: Delete the EKM connection.
+"""
+
+helps['keyvault ekm-connection certificate'] = """
+type: group
+short-summary: Manage EKM proxy certificate information.
+"""
+
+helps['keyvault ekm-connection certificate show'] = """
+type: command
+short-summary: Show the EKM proxy client certificate.
+"""
+
 helps['keyvault security-domain'] = """
 type: group
 short-summary: Manage security domain operations.

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

@@ -21,7 +21,7 @@
 from azure.cli.command_modules.keyvault._validators import (
     datetime_type, certificate_type, validate_retention_days_on_creation,
     get_vault_base_url_type, get_hsm_base_url_type, validate_key_import_type,
-    validate_key_import_source, validate_key_type, validate_policy_permissions, validate_principal,
+    validate_key_import_source, validate_policy_permissions, validate_principal,
     validate_resource_group_name, validate_x509_certificate_chain,
     secret_text_encoding_values, secret_binary_encoding_values, validate_subnet, validate_ip_address,
     validate_vault_or_hsm,
@@ -356,10 +356,13 @@ class CLISecurityDomainOperation(str, Enum):
                          'Release policies are mutable by default.')
 
     with self.argument_context('keyvault key create') as c:
-        c.argument('kty', arg_type=get_enum_type(JsonWebKeyType), validator=validate_key_type,
+        c.argument('kty', arg_type=get_enum_type(JsonWebKeyType),
                    help='The type of key to create. For valid values, see: https://learn.microsoft.com/rest/api/keyvault/keys/create-key/create-key#jsonwebkeytype')
         c.argument('curve', arg_type=get_enum_type(KeyCurveName),
                    help='Elliptic curve name. For valid values, see: https://learn.microsoft.com/rest/api/keyvault/keys/create-key/create-key#jsonwebkeycurvename')
+        c.argument('external_key_id', options_list=['--external-key-id'], arg_group='External Key',
+                   is_preview=True,
+                   help='Create an external Managed HSM key backed by an External Key Manager (EKM) key id.')
 
     with self.argument_context('keyvault key import') as c:
         c.argument('kty', arg_type=get_enum_type(CLIKeyTypeForBYOKImport), validator=validate_key_import_type,
@@ -616,6 +619,49 @@ class CLISecurityDomainOperation(str, Enum):
                    help='Target operation that needs waiting.')
     # endregion
 
+    # region keyvault ekm-connection
+    for scope in ['create', 'update', 'show', 'check', 'delete']:
+        with self.argument_context('keyvault ekm-connection {}'.format(scope), arg_group='HSM Id') as c:
+            c.extra('hsm_name', hsm_url_type, required=False,
+                    help='Name of the HSM. Can be omitted if --id is specified.')
+            c.extra('identifier', options_list=['--id'], validator=validate_vault_or_hsm,
+                    help='Full URI of the HSM.')
+            c.ignore('vault_base_url')
+
+    with self.argument_context('keyvault ekm-connection create', arg_group='EKM Connection') as c:
+        c.argument('host', options_list=['--host'], required=True,
+                   help='EKM proxy host (FQDN or FQDN:port). If port is omitted, 443 is assumed.')
+        c.extra('path_prefix', options_list=['--path-prefix'],
+                help='Optional path prefix to append to EKM proxy requests. Must start with "/".')
+        c.extra('server_ca_certificates', options_list=['--server-ca-certificate'], nargs='+', type=file_type,
+                required=True, completer=FilesCompleter(),
+                help='Path(s) to server CA certificate(s) in PEM or DER format. '
+                     'Pass a single file containing a PEM chain (multiple certificate blocks), '
+                     'or multiple space-separated file paths (each PEM or DER).')
+        c.extra('server_subject_common_name', options_list=['--server-subject-common-name', '--server-cn'],
+                help='Optional expected Common Name (CN) for the EKM proxy server certificate.')
+
+    with self.argument_context('keyvault ekm-connection update', arg_group='EKM Connection') as c:
+        c.argument('host', options_list=['--host'], required=False,
+                   help='EKM proxy host (FQDN or FQDN:port). If port is omitted, 443 is assumed.')
+        c.extra('path_prefix', options_list=['--path-prefix'],
+                help='Optional path prefix to append to EKM proxy requests. Must start with "/".')
+        c.extra('server_ca_certificates', options_list=['--server-ca-certificate'], nargs='+', type=file_type,
+                completer=FilesCompleter(),
+                help='Path(s) to server CA certificate(s) in PEM or DER format. '
+                     'Pass a single file containing a PEM chain (multiple certificate blocks), '
+                     'or multiple space-separated file paths (each PEM or DER).')
+        c.extra('server_subject_common_name', options_list=['--server-subject-common-name', '--server-cn'],
+                help='Optional expected Common Name (CN) for the EKM proxy server certificate.')
+
+    with self.argument_context('keyvault ekm-connection certificate show', arg_group='HSM Id') as c:
+        c.extra('hsm_name', hsm_url_type, required=False,
+                help='Name of the HSM. Can be omitted if --id is specified.')
+        c.extra('identifier', options_list=['--id'], validator=validate_vault_or_hsm,
+                help='Full URI of the HSM.')
+        c.ignore('vault_base_url')
+    # endregion
+
     # region keyvault backup/restore
     for item in ['backup', 'restore']:
         for scope in ['start']:  # TODO add 'status' when SDK is ready

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

@@ -60,9 +60,9 @@ def transform_key_encryption_output(result, **command_args):  # pylint: disable=
         'kid': result.key_id,
         'result': base64.b64encode(result.ciphertext).decode('utf-8'),
         'algorithm': result.algorithm,
-        'iv': binascii.hexlify(result.iv) if result.iv else None,
-        'tag': binascii.hexlify(result.tag) if result.tag else None,
-        'aad': binascii.hexlify(result.aad) if result.aad else None
+        'iv': binascii.hexlify(result.iv).decode('ascii') if result.iv else None,
+        'tag': binascii.hexlify(result.tag).decode('ascii') if result.tag else None,
+        'aad': binascii.hexlify(result.aad).decode('ascii') if result.aad else None
     }
     return output
 
@@ -105,6 +105,13 @@ def transform_key_list_output(result, **command_args):  # pylint: disable=unused
         k['managed'] = key.managed
         k['tags'] = key.tags
         k['releasePolicy'] = key.release_policy
+
+        # External key (EKM) is a preview property and may not exist on all SDK versions.
+        external_key = getattr(key, 'external_key', None)
+        external_key_id = getattr(external_key, 'id', None) if external_key else None
+        if external_key_id:
+            k['externalKeyId'] = external_key_id
+
         output.append(k)
     return output
 
@@ -141,6 +148,13 @@ def transform_key_output(result, **command_args):
         'tags': result.properties.tags,
         'releasePolicy': result.properties.release_policy
     }
+
+    # External key (EKM) is a preview property and may not exist on all SDK versions.
+    external_key = getattr(result.properties, 'external_key', None)
+    external_key_id = getattr(external_key, 'id', None) if external_key else None
+    if external_key_id:
+        output['externalKeyId'] = external_key_id
+
     if isinstance(result, DeletedKey):
         output['deletedDate'] = result.deleted_date
         output['scheduledPurgeDate'] = result.scheduled_purge_date

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

@@ -732,10 +732,163 @@ def validate_key_create(cmd, ns):
     validate_tags(ns)
     set_vault_base_url(ns)
     validate_keyvault_resource_id('key')(ns)
-    validate_key_type(ns)
+    validate_external_key_id(ns)
+
+    if getattr(ns, 'external_key_id', None):
+        # External keys are backed by an External Key Manager (EKM); the service controls the
+        # key material, so client-specified key-shape arguments are not supported. Fail fast with
+        # a clear error instead of silently ignoring them.
+        incompatible = [opt for opt, val in (
+            ('--kty', getattr(ns, 'kty', None)),
+            ('--size', getattr(ns, 'key_size', None)),
+            ('--curve', getattr(ns, 'curve', None)),
+            ('--ops', getattr(ns, 'key_ops', None)),
+            ('--protection', getattr(ns, 'protection', None)),
+            ('--exportable', getattr(ns, 'exportable', None)),
+        ) if val is not None]
+        if incompatible:
+            raise CLIError(
+                '{} cannot be used with --external-key-id. External keys are backed by an External '
+                'Key Manager and the service controls the key material.'.format(', '.join(incompatible)))
+    else:
+        validate_key_type(ns)
+
     process_key_release_policy(cmd, ns)
 
 
+def validate_external_key_id(ns):
+    external_key_id = getattr(ns, 'external_key_id', None)
+    if not external_key_id:
+        return
+    if len(external_key_id) > 64:
+        raise CLIError('--external-key-id must be at most 64 characters.')
+    if not re.match(r'^[0-9A-Za-z-]+$', external_key_id):
+        raise CLIError('--external-key-id may contain only letters, digits, and hyphens.')
+
+
+def _validate_ekm_path_prefix(path_prefix=None):
+    if path_prefix is None:
+        return
+    if not path_prefix.startswith('/'):
+        raise CLIError('--path-prefix must start with "/".')
+    if path_prefix.endswith('/'):
+        raise CLIError('--path-prefix must not end with "/".')
+    if len(path_prefix) > 64:
+        raise CLIError('--path-prefix must be at most 64 characters.')
+    if not re.match(r'^[A-Za-z0-9/-]+$', path_prefix):
+        raise CLIError('--path-prefix may contain only letters, digits, "/" and "-".')
+
+
+def _normalize_ekm_host(host: str):
+    host = (host or '').strip()
+    if not host:
+        raise CLIError('--host cannot be empty.')
+    if '://' in host:
+        raise CLIError('--host must not include a URL scheme (use FQDN or FQDN:port).')
+    if '/' in host:
+        raise CLIError('--host must not include a path (use FQDN or FQDN:port).')
+
+    if ':' not in host:
+        return f'{host}:443'
+
+    # Avoid ambiguous parsing for IPv6 literals.
+    if host.count(':') != 1:
+        raise CLIError('--host must be in the form FQDN or FQDN:port.')
+
+    hostname, port_str = host.split(':', 1)
+    if not hostname:
+        raise CLIError('--host must be in the form FQDN or FQDN:port.')
+    try:
+        port = int(port_str)
+    except ValueError as ex:
+        raise CLIError('--host port must be an integer.') from ex
+    if port < 1 or port > 65535:
+        raise CLIError('--host port must be between 1 and 65535.')
+    return f'{hostname}:{port}'
+
+
+def _flatten_list(value):
+    if value is None:
+        return None
+    if isinstance(value, list) and value and isinstance(value[0], list):
+        flattened = []
+        for item in value:
+            flattened.extend(item)
+        return flattened
+    return value
+
+
+def _load_certificates_as_der_bytes(cert_paths):
+    import os
+    import ssl
+
+    cert_paths = _flatten_list(cert_paths)
+    if not cert_paths:
+        return []
+
+    der_certs = []
+    for cert_path in cert_paths:
+        if not cert_path:
+            continue
+        expanded = os.path.expanduser(cert_path)
+        try:
+            with open(expanded, 'rb') as f:
+                raw = f.read()
+        except OSError as ex:
+            raise CLIError("Unable to load certificate file '{}': {}.".format(cert_path, ex.strerror)) from ex
+
+        # PEM may contain multiple cert blocks.
+        if b'-----BEGIN CERTIFICATE-----' in raw:
+            text = raw.decode('utf-8', errors='ignore')
+            begin = '-----BEGIN CERTIFICATE-----'
+            end = '-----END CERTIFICATE-----'
+            start = 0
+            found_any = False
+            while True:
+                b_idx = text.find(begin, start)
+                if b_idx == -1:
+                    break
+                e_idx = text.find(end, b_idx)
+                if e_idx == -1:
+                    raise CLIError(f'Invalid PEM certificate in {cert_path}.')
+                block = text[b_idx:e_idx + len(end)]
+                der_certs.append(ssl.PEM_cert_to_DER_cert(block))
+                found_any = True
+                start = e_idx + len(end)
+            if not found_any:
+                raise CLIError(f'Invalid PEM certificate in {cert_path}.')
+        else:
+            # Assume DER.
+            der_certs.append(raw)
+
+    return der_certs
+
+
+def validate_ekm_connection_base(cmd, ns):  # pylint: disable=unused-argument
+    set_vault_base_url(ns)
+    if not getattr(ns, 'hsm_name', None) and not getattr(ns, 'identifier', None):
+        raise CLIError('Please specify --hsm-name or --id.')
+
+
+def validate_ekm_connection_create(cmd, ns):
+    validate_ekm_connection_base(cmd, ns)
+    ns.host = _normalize_ekm_host(ns.host)
+    _validate_ekm_path_prefix(getattr(ns, 'path_prefix', None))
+    server_ca_certificates = _load_certificates_as_der_bytes(getattr(ns, 'server_ca_certificates', None))
+    if not server_ca_certificates:
+        raise CLIError('Please specify at least one --server-ca-certificate for EKM connection creation.')
+    ns.server_ca_certificates = server_ca_certificates
+
+
+def validate_ekm_connection_update(cmd, ns):
+    validate_ekm_connection_base(cmd, ns)
+    if getattr(ns, 'host', None):
+        ns.host = _normalize_ekm_host(ns.host)
+    _validate_ekm_path_prefix(getattr(ns, 'path_prefix', None))
+    if getattr(ns, 'server_ca_certificates', None):
+        ns.server_ca_certificates = _load_certificates_as_der_bytes(ns.server_ca_certificates)
+
+
 # pylint: disable=line-too-long, too-many-locals
 def process_certificate_policy(cmd, ns):
     policy = getattr(ns, 'policy', None)