Update CLI to create role assignment (Azure#8636)

* create role assignment for Fleet RP on subnet when creating private fleet

* lint

* update

* update

* update

* comment

* rename func

Author: MuhammadAliFleet

src/fleet/HISTORY.rst

@@ -118,4 +118,8 @@ Release History
 
 1.5.0
 ++++++
-* Upgrade SDK version to 2025-03-01
+* Upgrade SDK version to 2025-03-01
+
+1.5.1
+++++++
+* create_fleet now creates a role assignment when fleet type is private

src/fleet/azext_fleet/_client_factory.py

@@ -44,3 +44,8 @@ def cf_auto_upgrade_profiles(cli_ctx, *_):
 
 def cf_auto_upgrade_profile_operations(cli_ctx, *_):
     return get_container_service_client(cli_ctx).auto_upgrade_profile_operations
+
+
+def get_provider_client(cli_ctx):
+    return get_mgmt_service_client(
+        cli_ctx, ResourceType.MGMT_RESOURCE_RESOURCES)

src/fleet/azext_fleet/_helpers.py

@@ -13,6 +13,10 @@
 from knack.log import get_logger
 from knack.prompting import NoTTYException, prompt_y_n
 from knack.util import CLIError
+from azure.cli.command_modules.acs._roleassignments import add_role_assignment
+
+from azext_fleet.constants import FLEET_1P_APP_ID
+from azext_fleet._client_factory import get_provider_client
 
 logger = get_logger(__name__)
 
@@ -148,3 +152,17 @@ def _load_kubernetes_configuration(filename):
         raise
     except (yaml.parser.ParserError, UnicodeDecodeError) as ex:
         raise CLIError(f'Error parsing {filename} ({str(ex)})') from ex
+
+
+def assign_network_contributor_role_to_subnet(cmd, subnet_id):
+    resource_client = get_provider_client(cmd.cli_ctx)
+    provider = resource_client.providers.get("Microsoft.ContainerService")
+
+    # provider registration state being is checked to ensure that the Fleet service principal is available
+    # to create the role assignment on the subnet
+    if provider.registration_state != 'Registered':
+        raise CLIError("The Microsoft.ContainerService resource provider is not registered."
+                       "Run `az provider register -n Microsoft.ContainerService --wait`.")
+    if not add_role_assignment(cmd, 'Network Contributor', FLEET_1P_APP_ID, scope=subnet_id):
+        raise CLIError("failed to create role assignment for Fleet RP.\n"
+                       f"Do you have owner permissions on the subnet {subnet_id}?\n")

src/fleet/azext_fleet/constants.py

@@ -6,6 +6,7 @@
 UPGRADE_TYPE_CONTROLPLANEONLY = "ControlPlaneOnly"
 UPGRADE_TYPE_FULL = "Full"
 UPGRADE_TYPE_NODEIMAGEONLY = "NodeImageOnly"
+FLEET_1P_APP_ID = "609d2f62-527f-4451-bfd2-ac2c7850822c"
 
 UPGRADE_TYPE_ERROR_MESSAGES = {
     UPGRADE_TYPE_CONTROLPLANEONLY: f"Please set kubernetes version when upgrade type is '{UPGRADE_TYPE_CONTROLPLANEONLY}'.",  # pylint: disable=line-too-long

src/fleet/azext_fleet/custom.py

@@ -13,6 +13,7 @@
 
 from azext_fleet._client_factory import CUSTOM_MGMT_FLEET
 from azext_fleet._helpers import print_or_merge_credentials
+from azext_fleet._helpers import assign_network_contributor_role_to_subnet
 from azext_fleet.constants import UPGRADE_TYPE_CONTROLPLANEONLY
 from azext_fleet.constants import UPGRADE_TYPE_FULL
 from azext_fleet.constants import UPGRADE_TYPE_NODEIMAGEONLY
@@ -109,6 +110,9 @@ def create_fleet(cmd,
         identity=managed_service_identity
     )
 
+    if enable_private_cluster:
+        assign_network_contributor_role_to_subnet(cmd, resource_group_name, agent_subnet_id)
+
     return sdk_no_wait(no_wait,
                        client.begin_create_or_update,
                        resource_group_name,

src/fleet/azext_fleet/tests/latest/recordings/test_fleet_hubful.yaml

@@ -16,7 +16,7 @@
 
 # TODO: Confirm this is the right version number you want and it matches your
 # HISTORY.rst entry.
-VERSION = '1.5.0'
+VERSION = '1.5.1'
 
 # The full list of classifiers is available at
 # https://pypi.python.org/pypi?%3Aaction=list_classifiers