[Key Vault] az keyvault ekm-connection: Add private endpoint support

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

@@ -260,6 +260,7 @@ def data_plane_azure_keyvault_ekm_client(cli_ctx, command_args):
     return KeyVaultEkmClient(
         vault_url=vault_url,
         credential=credential,
+        api_version='2026-07-01-preview',
         verify_challenge_resource=False,
         **client_kwargs)
 

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

@@ -983,11 +983,23 @@
 helps['keyvault ekm-connection create'] = """
 type: command
 short-summary: Create the EKM connection.
+long-summary: In PrivateEndpoint mode, host is the name of an EKM private endpoint whose connection has been approved by the Private Link Service owner. The proxy CA certificates are required in both modes.
+examples:
+  - name: Connect to an EKM proxy over the public network.
+    text: az keyvault ekm-connection create --hsm-name MyHSM --host proxy.example.com --path-prefix /api/v1 --server-ca-certificate proxy-ca.pem
+  - name: Connect to an approved EKM private endpoint.
+    text: az keyvault ekm-connection create --hsm-name MyHSM --host ekm-proxy-pe --connectivity-mode PrivateEndpoint --path-prefix /api/v1 --server-ca-certificate proxy-ca.pem --server-cn proxy.example.com
 """
 
 helps['keyvault ekm-connection update'] = """
 type: command
 short-summary: Update the EKM connection.
+long-summary: Omitted fields retain their current values. Specify host when changing connectivity mode.
+examples:
+  - name: Switch an EKM connection to an approved private endpoint.
+    text: az keyvault ekm-connection update --hsm-name MyHSM --connectivity-mode PrivateEndpoint --host ekm-proxy-pe
+  - name: Switch an EKM connection back to a public proxy.
+    text: az keyvault ekm-connection update --hsm-name MyHSM --connectivity-mode Public --host proxy.example.com:443
 """
 
 helps['keyvault ekm-connection show'] = """
@@ -1015,6 +1027,80 @@
 short-summary: Show the EKM proxy client certificate.
 """
 
+helps['keyvault ekm-connection private-endpoint'] = """
+type: group
+short-summary: Manage private endpoints from a Managed HSM to an EKM proxy.
+long-summary: These are outbound EKM proxy endpoints, not inbound Managed HSM private endpoint connections. A Managed HSM supports up to two EKM private endpoints.
+"""
+
+helps['keyvault ekm-connection private-endpoint create'] = """
+type: command
+short-summary: Create an EKM proxy private endpoint.
+long-summary: Returns the completed operation record unless no-wait is specified. The Private Link Service owner must approve the connection before it can be used by an EKM connection.
+examples:
+  - name: Create an endpoint using a Private Link Service alias.
+    text: az keyvault ekm-connection private-endpoint create --hsm-name MyHSM --name ekm-proxy-pe --private-link-service MyService.Alias
+  - name: Request an endpoint without waiting for provisioning.
+    text: az keyvault ekm-connection private-endpoint create --id https://MyHSM.managedhsm.azure.net --name ekm-proxy-pe --private-link-service MyService.Alias --request-message "Please approve this connection" --no-wait
+"""
+
+helps['keyvault ekm-connection private-endpoint delete'] = """
+type: command
+short-summary: Delete an EKM proxy private endpoint.
+long-summary: Deletion is rejected while an EKM connection references the endpoint. Update or remove that connection first. Returns the completed operation record unless no-wait is specified.
+examples:
+  - name: Delete an unused EKM private endpoint.
+    text: az keyvault ekm-connection private-endpoint delete --hsm-name MyHSM --name ekm-proxy-pe
+  - name: Start deletion without prompting or waiting.
+    text: az keyvault ekm-connection private-endpoint delete --id https://MyHSM.managedhsm.azure.net --name ekm-proxy-pe --yes --no-wait
+"""
+
+helps['keyvault ekm-connection private-endpoint show'] = """
+type: command
+short-summary: Show an EKM private endpoint and its connection approval status.
+examples:
+  - name: Show an EKM private endpoint.
+    text: az keyvault ekm-connection private-endpoint show --hsm-name MyHSM --name ekm-proxy-pe
+  - name: Read the Private Link Service approval status.
+    text: az keyvault ekm-connection private-endpoint show --id https://MyHSM.managedhsm.azure.net --name ekm-proxy-pe --query privateLinkServiceConnectionState.status -o tsv
+"""
+
+helps['keyvault ekm-connection private-endpoint list'] = """
+type: command
+short-summary: List EKM private endpoints on a Managed HSM.
+examples:
+  - name: List all EKM private endpoints.
+    text: az keyvault ekm-connection private-endpoint list --hsm-name MyHSM
+  - name: List the names of approved endpoints.
+    text: az keyvault ekm-connection private-endpoint list --id https://MyHSM.managedhsm.azure.net --query "[?privateLinkServiceConnectionState.status=='Approved'].name" -o tsv
+"""
+
+helps['keyvault ekm-connection private-endpoint wait'] = """
+type: command
+short-summary: Wait for an EKM private endpoint to reach a condition.
+long-summary: The created condition waits for provisioning to succeed, not for Private Link Service approval. Use a custom condition to wait for approval.
+examples:
+  - name: Wait for endpoint provisioning to complete.
+    text: az keyvault ekm-connection private-endpoint wait --hsm-name MyHSM --name ekm-proxy-pe --created
+  - name: Wait for the Private Link Service owner to approve the connection.
+    text: az keyvault ekm-connection private-endpoint wait --hsm-name MyHSM --name ekm-proxy-pe --custom "privateLinkServiceConnectionState.status=='Approved'" --interval 10 --timeout 600
+"""
+
+helps['keyvault ekm-connection private-endpoint operation'] = """
+type: group
+short-summary: Inspect EKM private endpoint operations.
+"""
+
+helps['keyvault ekm-connection private-endpoint operation show'] = """
+type: command
+short-summary: Show the status of an EKM private endpoint create or delete operation.
+examples:
+  - name: Show a private endpoint operation.
+    text: az keyvault ekm-connection private-endpoint operation show --hsm-name MyHSM --job-id 00000000-0000-0000-0000-000000000000
+  - name: Read the status of an operation without waiting for it to complete.
+    text: az keyvault ekm-connection private-endpoint operation show --id https://MyHSM.managedhsm.azure.net --job-id 00000000-0000-0000-0000-000000000000 --query status -o tsv
+"""
+
 helps['keyvault security-domain'] = """
 type: group
 short-summary: Manage security domain operations.

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

@@ -630,7 +630,10 @@ class CLISecurityDomainOperation(str, Enum):
 
     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.')
+                   help='EKM proxy host (FQDN or FQDN:port), or the private endpoint name in PrivateEndpoint mode. '
+                        'For Public mode, port 443 is assumed if omitted.')
+        c.argument('connectivity_mode', arg_type=get_enum_type(['Public', 'PrivateEndpoint']), is_preview=True,
+                   help='Connectivity mode for the EKM proxy. The service defaults to Public when omitted.')
         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,
@@ -643,7 +646,10 @@ class CLISecurityDomainOperation(str, Enum):
 
     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.')
+                   help='EKM proxy host (FQDN or FQDN:port), or the private endpoint name in PrivateEndpoint mode. '
+                        'Required when changing connectivity mode.')
+        c.argument('connectivity_mode', arg_type=get_enum_type(['Public', 'PrivateEndpoint']), is_preview=True,
+                   help='Connectivity mode for the EKM proxy. Preserves the existing mode when omitted.')
         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,
@@ -660,6 +666,28 @@ class CLISecurityDomainOperation(str, Enum):
         c.extra('identifier', options_list=['--id'], validator=validate_vault_or_hsm,
                 help='Full URI of the HSM.')
         c.ignore('vault_base_url')
+
+    for scope in ['create', 'delete', 'show', 'list', 'wait', 'operation show']:
+        with self.argument_context('keyvault ekm-connection private-endpoint {}'.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 private-endpoint') as c:
+        c.argument('private_endpoint_name', options_list=['--name', '-n'],
+                   help='EKM private endpoint name. Use 1-24 letters, digits or hyphens, '
+                        'starting and ending with a letter or digit.')
+
+    with self.argument_context('keyvault ekm-connection private-endpoint create') as c:
+        c.argument('private_link_service_id', options_list=['--private-link-service'],
+                   help='Alias of the Private Link Service that fronts the EKM proxy.')
+        c.argument('request_message', help='Message for the Private Link Service owner when approving the connection.')
+
+    with self.argument_context('keyvault ekm-connection private-endpoint operation show') as c:
+        c.argument('job_id', help='Job ID returned by a private endpoint create or delete operation.')
     # endregion
 
     # region keyvault backup/restore

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

@@ -807,6 +807,21 @@ def _normalize_ekm_host(host: str):
     return f'{hostname}:{port}'
 
 
+def _validate_ekm_private_endpoint_name(name):
+    if not isinstance(name, str) or not re.fullmatch(r'[A-Za-z0-9](?:[A-Za-z0-9-]{0,22}[A-Za-z0-9])?', name):
+        raise InvalidArgumentValueError('The EKM private endpoint name must be 1-24 letters, digits or hyphens, '
+                                        'and start and end with a letter or digit.')
+    return name
+
+
+def _normalize_ekm_connection_host(host, connectivity_mode=None):
+    if connectivity_mode == 'PrivateEndpoint':
+        return _validate_ekm_private_endpoint_name(host)
+    if connectivity_mode not in (None, 'Public'):
+        raise InvalidArgumentValueError('--connectivity-mode must be Public or PrivateEndpoint.')
+    return _normalize_ekm_host(host)
+
+
 def _flatten_list(value):
     if value is None:
         return None
@@ -872,7 +887,7 @@ def validate_ekm_connection_base(cmd, ns):  # pylint: disable=unused-argument
 
 def validate_ekm_connection_create(cmd, ns):
     validate_ekm_connection_base(cmd, ns)
-    ns.host = _normalize_ekm_host(ns.host)
+    ns.host = _normalize_ekm_connection_host(ns.host, getattr(ns, 'connectivity_mode', None))
     _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:
@@ -882,13 +897,19 @@ def validate_ekm_connection_create(cmd, ns):
 
 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)
+    if getattr(ns, 'host', None) is not None and getattr(ns, 'connectivity_mode', None) is not None:
+        ns.host = _normalize_ekm_connection_host(ns.host, ns.connectivity_mode)
     _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)
 
 
+def validate_ekm_private_endpoint(cmd, ns):
+    validate_ekm_connection_base(cmd, ns)
+    if getattr(ns, 'private_endpoint_name', None) is not None:
+        _validate_ekm_private_endpoint_name(ns.private_endpoint_name)
+
+
 # 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

@@ -27,7 +27,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_ekm_connection_base, validate_ekm_connection_create, validate_ekm_connection_update)
+    validate_ekm_connection_base, validate_ekm_connection_create, validate_ekm_connection_update,
+    validate_ekm_private_endpoint)
 
 
 def transform_assignment_list(result):
@@ -153,6 +154,21 @@ def load_command_table(self, _):
     with self.command_group('keyvault ekm-connection certificate', command_type=data_ekm_custom, is_preview=True) as g:
         g.keyvault_custom('show', 'get_ekm_certificate', validator=validate_ekm_connection_base)
 
+    with self.command_group('keyvault ekm-connection private-endpoint', command_type=data_ekm_custom,
+                            is_preview=True) as g:
+        g.keyvault_custom('create', 'create_ekm_private_endpoint', validator=validate_ekm_private_endpoint,
+                          supports_no_wait=True)
+        g.keyvault_custom('delete', 'delete_ekm_private_endpoint', validator=validate_ekm_private_endpoint,
+                          supports_no_wait=True, confirmation=True)
+        g.keyvault_custom('show', 'get_ekm_private_endpoint', validator=validate_ekm_private_endpoint)
+        g.keyvault_custom('list', 'list_ekm_private_endpoints', validator=validate_ekm_connection_base)
+        g.custom_wait_command('wait', 'get_ekm_private_endpoint', validator=validate_ekm_private_endpoint,
+                              custom_command_type=data_ekm_custom)
+
+    with self.command_group('keyvault ekm-connection private-endpoint operation', command_type=data_ekm_custom,
+                            is_preview=True) as g:
+        g.keyvault_custom('show', 'get_ekm_private_endpoint_operation', 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)

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

@@ -1254,30 +1254,70 @@ def delete_ekm_connection(client):
     return client.delete_ekm_connection()
 
 
-def create_ekm_connection(client, host, path_prefix=None, server_ca_certificates=None, server_subject_common_name=None):
+def create_ekm_connection(client, host, path_prefix=None, server_ca_certificates=None, server_subject_common_name=None,
+                          connectivity_mode=None):
     from azure.keyvault.administration import KeyVaultEkmConnection
 
     ekm_connection = KeyVaultEkmConnection(
         host=host,
         path_prefix=path_prefix,
         server_ca_certificates=server_ca_certificates,
-        server_subject_common_name=server_subject_common_name
+        server_subject_common_name=server_subject_common_name,
+        connectivity_mode=connectivity_mode
     )
     return client.create_ekm_connection(ekm_connection)
 
 
 def update_ekm_connection(client, host=None, path_prefix=None, server_ca_certificates=None,
-                          server_subject_common_name=None):
+                          server_subject_common_name=None, connectivity_mode=None):
+    from azure.cli.command_modules.keyvault._validators import _normalize_ekm_connection_host
+
     existing = client.get_ekm_connection()
+    existing_mode = getattr(existing, 'connectivity_mode', None) or 'Public'
+    if connectivity_mode is not None and connectivity_mode != existing_mode and host is None:
+        raise RequiredArgumentMissingError('--host is required when changing --connectivity-mode.')
     if host is not None:
-        existing.host = host
+        existing.host = _normalize_ekm_connection_host(host, connectivity_mode or existing_mode)
+    if connectivity_mode is not None:
+        existing.connectivity_mode = connectivity_mode
     if path_prefix is not None:
         existing.path_prefix = path_prefix
     if server_ca_certificates is not None:
         existing.server_ca_certificates = server_ca_certificates
     if server_subject_common_name is not None:
         existing.server_subject_common_name = server_subject_common_name
     return client.update_ekm_connection(existing)
+
+
+def create_ekm_private_endpoint(client, private_endpoint_name, private_link_service_id, request_message=None,
+                                no_wait=False):
+    from azure.mgmt.core.polling.arm_polling import ARMPolling
+
+    polling = False if no_wait else ARMPolling(lro_options={'final-state-via': 'azure-async-operation'})
+    poller = client.begin_create_ekm_private_endpoint(
+        name=private_endpoint_name, private_link_service_id=private_link_service_id,
+        request_message=request_message, polling=polling)
+    return None if no_wait else poller
+
+
+def delete_ekm_private_endpoint(client, private_endpoint_name, no_wait=False):
+    from azure.mgmt.core.polling.arm_polling import ARMPolling
+
+    polling = False if no_wait else ARMPolling(lro_options={'final-state-via': 'azure-async-operation'})
+    poller = client.begin_delete_ekm_private_endpoint(name=private_endpoint_name, polling=polling)
+    return None if no_wait else poller
+
+
+def get_ekm_private_endpoint(client, private_endpoint_name):
+    return client.get_ekm_private_endpoint(name=private_endpoint_name)
+
+
+def list_ekm_private_endpoints(client):
+    return list(client.list_ekm_private_endpoints())
+
+
+def get_ekm_private_endpoint_operation(client, job_id):
+    return client.get_ekm_private_endpoint_operation_status(job_id=job_id)
 # endregion