[AKS] az aks identity-binding: Add command group to manage identity bindings (trust domain) for a managed cluster (#33558)

Author: toma3233

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

@@ -39,6 +39,10 @@ def cf_managed_namespaces(cli_ctx, *_):
     return get_container_service_client(cli_ctx).managed_namespaces
 
 
+def cf_identity_bindings(cli_ctx, *_):
+    return get_container_service_client(cli_ctx).identity_bindings
+
+
 def cf_agent_pools(cli_ctx, *_):
     return get_container_service_client(cli_ctx).agent_pools
 

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

@@ -2832,6 +2832,71 @@
           short-summary: Specify the role binding name.
 """
 
+helps["aks identity-binding"] = """
+    type: group
+    short-summary: Commands to manage identity bindings in Azure Kubernetes Service.
+"""
+
+helps["aks identity-binding list"] = """
+    type: command
+    short-summary: List all identity bindings under a managed Kubernetes cluster.
+    parameters:
+        - name: --cluster-name
+          type: string
+          short-summary: Name of the managed Kubernetes cluster.
+    examples:
+        - name: List all identity bindings in a managed cluster
+          text: az aks identity-binding list -g myResourceGroup --cluster-name myCluster
+"""
+
+helps["aks identity-binding show"] = """
+    type: command
+    short-summary: Show details of a specific identity binding in a managed Kubernetes cluster.
+    parameters:
+        - name: --cluster-name
+          type: string
+          short-summary: Name of the managed Kubernetes cluster.
+        - name: --name -n
+          type: string
+          short-summary: Name of the identity binding to show.
+    examples:
+        - name: Show details of an identity binding
+          text: az aks identity-binding show -g myResourceGroup --cluster-name myCluster -n myIdentityBinding
+"""
+
+helps["aks identity-binding create"] = """
+    type: command
+    short-summary: Create a new identity binding in a managed Kubernetes cluster.
+    parameters:
+        - name: --cluster-name
+          type: string
+          short-summary: Name of the managed Kubernetes cluster.
+        - name: --name -n
+          type: string
+          short-summary: Name of the identity binding to create.
+        - name: --managed-identity-resource-id
+          type: string
+          short-summary: The resource ID of the managed identity to use.
+    examples:
+        - name: Create a new identity binding
+          text: az aks identity-binding create -g myResourceGroup --cluster-name myCluster -n myIdentityBinding --managed-identity-resource-id /subscriptions/0000/resourceGroups/myResourceGroup/providers/Microsoft.ManagedIdentity/userAssignedIdentities/myIdentity
+"""
+
+helps["aks identity-binding delete"] = """
+    type: command
+    short-summary: Delete a specific identity binding in a managed Kubernetes cluster.
+    parameters:
+        - name: --cluster-name
+          type: string
+          short-summary: Name of the managed Kubernetes cluster.
+        - name: --name -n
+          type: string
+          short-summary: Name of the identity binding to delete.
+    examples:
+        - name: Delete an identity binding
+          text: az aks identity-binding delete -g myResourceGroup --cluster-name myCluster -n myIdentityBinding
+"""
+
 helps["aks mesh"] = """
     type: group
     short-summary: Commands to manage Azure Service Mesh.

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

@@ -1287,6 +1287,21 @@ def load_arguments(self, _):
     with self.argument_context('aks trustedaccess rolebinding update') as c:
         c.argument('roles', help='comma-separated roles: Microsoft.Demo/samples/reader,Microsoft.Demo/samples/writer,...')
 
+    with self.argument_context('aks identity-binding') as c:
+        c.argument('cluster_name', help='Name of the managed cluster.')
+
+    for scope in ['aks identity-binding show', 'aks identity-binding create', 'aks identity-binding delete']:
+        with self.argument_context(scope) as c:
+            c.argument('name', options_list=['--name', '-n'], required=True,
+                       help='Name of the identity binding.')
+
+    with self.argument_context('aks identity-binding create') as c:
+        c.argument(
+            'managed_identity_resource_id',
+            options_list=['--managed-identity-resource-id'],
+            help='The resource ID of the managed identity to use.',
+        )
+
     with self.argument_context('aks mesh enable-ingress-gateway') as c:
         c.argument('ingress_gateway_type',
                    arg_type=get_enum_type(ingress_gateway_types))

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

@@ -11,7 +11,8 @@
     cf_snapshots,
     cf_trustedaccess_role,
     cf_trustedaccess_role_binding,
-    cf_machines
+    cf_machines,
+    cf_identity_bindings
 )
 from azure.cli.command_modules.acs._format import (
     aks_agentpool_list_table_format,
@@ -100,6 +101,14 @@ def load_command_table(self, _):
         client_factory=cf_trustedaccess_role_binding
     )
 
+    identity_bindings_sdk = CliCommandType(
+        operations_tmpl='azure.mgmt.containerservice.operations.'
+                        '_operations#IdentityBindingsOperations.{}',
+        operation_group='identity_bindings',
+        resource_type=ResourceType.MGMT_CONTAINERSERVICE,
+        client_factory=cf_identity_bindings
+    )
+
     # AKS commands
     with self.command_group('aks', managed_clusters_sdk,
                             client_factory=cf_managed_clusters) as g:
@@ -266,6 +275,13 @@ def load_command_table(self, _):
         g.custom_command(
             'delete', 'aks_trustedaccess_role_binding_delete', confirmation=True)
 
+    # AKS identity binding commands
+    with self.command_group('aks identity-binding', identity_bindings_sdk, client_factory=cf_identity_bindings) as g:
+        g.custom_command('create', 'aks_identity_binding_create', supports_no_wait=True)
+        g.custom_command('delete', 'aks_identity_binding_delete', supports_no_wait=True, confirmation=True)
+        g.custom_show_command('show', 'aks_identity_binding_show')
+        g.custom_command('list', 'aks_identity_binding_list')
+
     # AKS mesh commands
     with self.command_group('aks mesh', managed_clusters_sdk, client_factory=cf_managed_clusters) as g:
         g.custom_command(

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

@@ -3802,6 +3802,50 @@ def aks_trustedaccess_role_binding_delete(cmd, client, resource_group_name, clus
     return client.begin_delete(resource_group_name, cluster_name, role_binding_name)
 
 
+def aks_identity_binding_create(cmd, client, resource_group_name, cluster_name, name,
+                                managed_identity_resource_id, no_wait=False):
+    IdentityBinding, IdentityBindingProperties, IdentityBindingManagedIdentityProfile = cmd.get_models(
+        "IdentityBinding",
+        "IdentityBindingProperties",
+        "IdentityBindingManagedIdentityProfile",
+        resource_type=ResourceType.MGMT_CONTAINERSERVICE,
+        operation_group="identity_bindings",
+    )
+    instance = IdentityBinding(
+        properties=IdentityBindingProperties(
+            managed_identity=IdentityBindingManagedIdentityProfile(
+                resource_id=managed_identity_resource_id,
+            )
+        )
+    )
+    return sdk_no_wait(
+        no_wait,
+        client.begin_create_or_update,
+        resource_group_name,
+        cluster_name,
+        name,
+        instance,
+    )
+
+
+def aks_identity_binding_delete(cmd, client, resource_group_name, cluster_name, name, no_wait=False):  # pylint: disable=unused-argument
+    return sdk_no_wait(
+        no_wait,
+        client.begin_delete,
+        resource_group_name,
+        cluster_name,
+        name,
+    )
+
+
+def aks_identity_binding_show(cmd, client, resource_group_name, cluster_name, name):  # pylint: disable=unused-argument
+    return client.get(resource_group_name, cluster_name, name)
+
+
+def aks_identity_binding_list(cmd, client, resource_group_name, cluster_name):  # pylint: disable=unused-argument
+    return client.list_by_managed_cluster(resource_group_name, cluster_name)
+
+
 def aks_mesh_enable(
         cmd,
         client,

src/azure-cli/azure/cli/command_modules/acs/linter_exclusions.yml

@@ -286,4 +286,9 @@ aks mesh enable:
       proxy_redirection_mechanism:
         rule_exclusions:
         - option_length_too_long
+aks identity-binding create:
+  parameters:
+      managed_identity_resource_id:
+        rule_exclusions:
+        - option_length_too_long
 ...

src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_aks_identity_binding.py

@@ -0,0 +1,111 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+
+from azure.cli.testsdk import ScenarioTest, live_only
+from azure.cli.testsdk.scenario_tests import AllowLargeResponse
+
+from azure.cli.command_modules.acs.tests.latest.recording_processors import KeyReplacer
+from azure.cli.command_modules.acs.tests.latest.custom_preparers import (
+    AKSCustomResourceGroupPreparer,
+)
+
+
+class IdentityBindingTestCases(ScenarioTest):
+
+    def __init__(self, method_name):
+        super(IdentityBindingTestCases, self).__init__(
+            method_name,
+            recording_processors=[KeyReplacer()],
+        )
+
+    @AllowLargeResponse()
+    @live_only()
+    @AKSCustomResourceGroupPreparer(
+        random_name_length=17,
+        name_prefix="clitest",
+        location="eastus2",
+    )
+    def test_identity_binding_usages(self, resource_group, resource_group_location):
+        aks_name = self.create_random_name("cliakstest", 16)
+        identity_name = self.create_random_name("cli", 16)
+        identity_binding_name = self.create_random_name("cliib", 16)
+        self.kwargs.update(
+            {
+                "resource_group": resource_group,
+                "aks_name": aks_name,
+                "identity_name": identity_name,
+                "identity_binding_name": identity_binding_name,
+                "location": resource_group_location,
+            }
+        )
+
+        create_aks_cmd = ("aks create --resource-group={resource_group} --name={aks_name} "
+                          "--location={location} --no-ssh-key -o json")
+        self.cmd(create_aks_cmd, checks=[
+                 self.check("provisioningState", "Succeeded")])
+
+        list_identity_binding_cmd = ("aks identity-binding list --resource-group {resource_group} "
+                                     "--cluster-name {aks_name} -o json")
+        self.cmd(
+            list_identity_binding_cmd,
+            checks=[
+                self.check("length(@)", 0)
+            ]
+        )
+
+        create_identity_cmd = ("identity create --resource-group {resource_group} --name {identity_name} "
+                               "--location {location} -o json")
+        identity = self.cmd(create_identity_cmd).get_output_in_json()
+        identity_resource_id = identity["id"]
+        identity_client_id = identity["clientId"]
+        identity_tenant_id = identity["tenantId"]
+
+        identity_binding_checks = [
+            self.check("properties.provisioningState", "Succeeded"),
+            self.check(
+                "properties.managedIdentity.resourceId",
+                identity_resource_id
+            ),
+            self.check(
+                "properties.managedIdentity.clientId",
+                identity_client_id
+            ),
+            self.check(
+                "properties.managedIdentity.tenantId",
+                identity_tenant_id
+            ),
+            self.check(
+                f"ends_with(properties.oidcIssuer.oidcIssuerUrl, '/{identity_tenant_id}/{identity_client_id}')",
+                True,
+            ),
+        ]
+
+        create_identity_binding_cmd = ("aks identity-binding create --resource-group {resource_group} --cluster-name {aks_name} "
+                                       "-n {identity_binding_name} -o json"
+                                       f" --managed-identity-resource-id {identity_resource_id}")
+        self.cmd(create_identity_binding_cmd, checks=identity_binding_checks)
+
+        self.cmd(
+            list_identity_binding_cmd,
+            checks=[
+                self.check("length(@)", 1)
+            ]
+        )
+
+        show_identity_binding_cmd = ("aks identity-binding show --resource-group {resource_group} --cluster-name {aks_name} "
+                                     "-n {identity_binding_name} -o json")
+        self.cmd(show_identity_binding_cmd, checks=identity_binding_checks)
+
+        delete_identity_binding_cmd = ("aks identity-binding delete --resource-group {resource_group} --cluster-name {aks_name} "
+                                       "-n {identity_binding_name} --yes -o json")
+        self.cmd(delete_identity_binding_cmd)
+
+        self.cmd(
+            list_identity_binding_cmd,
+            checks=[
+                self.check("length(@)", 0)
+            ]
+        )