Yash/ekm support for hsm

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

@@ -360,6 +360,8 @@ class CLISecurityDomainOperation(str, Enum):
                    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.extra('external_key_id', options_list=['--external-key-id'], arg_group='External Key',
+            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 +618,45 @@ 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,
+                completer=FilesCompleter(),
+                help='Path(s) to server CA certificate(s) in PEM or DER format.')
+        c.extra('server_subject_common_name', options_list=['--server-subject-common-name'],
+                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.')
+        c.extra('server_subject_common_name', options_list=['--server-subject-common-name'],
+                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
 
@@ -104,6 +104,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
 
@@ -139,6 +146,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,151 @@ 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)
+
+    # External keys are backed by EKM and the service rejects client-specified key type/size/curve.
+    # Avoid the defaulting behavior in validate_key_type (RSA) when --external-key-id is present.
+    if getattr(ns, 'external_key_id', None):
+        setattr(ns, 'kty', None)
+        setattr(ns, 'key_size', None)
+        setattr(ns, 'curve', None)
+        setattr(ns, 'protection', None)
+    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) > 128:
+        raise CLIError('--external-key-id must be at most 128 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)
+        with open(expanded, 'rb') as f:
+            raw = f.read()
+
+        # 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)

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

@@ -9,7 +9,7 @@
 from azure.cli.core.profiles import ResourceType
 
 from azure.cli.command_modules.keyvault._client_factory import (
-    get_client, get_client_factory, Clients)
+    get_client, get_client_factory, Clients, data_plane_azure_keyvault_ekm_client)
 
 from azure.cli.command_modules.keyvault._transformers import (
     filter_out_managed_resources,
@@ -26,7 +26,8 @@
 
 from azure.cli.command_modules.keyvault._validators import (
     process_secret_set_namespace, validate_key_create,
-    validate_private_endpoint_connection_id, validate_role_assignment_args)
+    validate_private_endpoint_connection_id, validate_role_assignment_args,
+    validate_ekm_connection_base, validate_ekm_connection_create, validate_ekm_connection_update)
 
 
 def transform_assignment_list(result):
@@ -65,6 +66,11 @@ def load_command_table(self, _):
         operations_tmpl='azure.cli.command_modules.keyvault.custom#{}',
         client_factory=get_client_factory(ResourceType.MGMT_KEYVAULT, Clients.managed_hsms)
     )
+
+    data_ekm_custom = CliCommandType(
+        operations_tmpl='azure.cli.command_modules.keyvault.custom#{}',
+        client_factory=data_plane_azure_keyvault_ekm_client
+    )
     # endregion
 
     # Management Plane Commands
@@ -137,6 +143,16 @@ def load_command_table(self, _):
         g.keyvault_custom('download', 'security_domain_download', supports_no_wait=True)
         g.keyvault_custom('wait', '_wait_security_domain_operation')
 
+    with self.command_group('keyvault ekm-connection', command_type=data_ekm_custom) as g:
+        g.keyvault_custom('create', 'create_ekm_connection', validator=validate_ekm_connection_create)
+        g.keyvault_custom('update', 'update_ekm_connection', validator=validate_ekm_connection_update)
+        g.keyvault_custom('show', 'get_ekm_connection', validator=validate_ekm_connection_base)
+        g.keyvault_custom('check', 'check_ekm_connection', validator=validate_ekm_connection_base)
+        g.keyvault_custom('delete', 'delete_ekm_connection', validator=validate_ekm_connection_base)
+
+    with self.command_group('keyvault ekm-connection certificate', command_type=data_ekm_custom) as g:
+        g.keyvault_custom('show', 'get_ekm_certificate', validator=validate_ekm_connection_base)
+
     with self.command_group('keyvault key', data_key_entity.command_type) as g:
         g.keyvault_custom('create', 'create_key', transform=transform_key_output, validator=validate_key_create)
         g.keyvault_command('set-attributes', 'update_key_properties', transform=transform_key_output)