Support blob storage token store authenticate with MSI

Author: njuCZ

src/containerapp/HISTORY.rst

@@ -4,6 +4,7 @@ Release History
 ===============
 upcoming
 ++++++
+* 'az containerapp auth update': Support authenticating blob storage token store using managed identity with `--blob-container-uri` and `--blob-container-identity`.
 
 1.1.0b4
 ++++++

src/containerapp/azext_containerapp/_help.py

@@ -715,6 +715,12 @@
   - name: Configure the app to listen to the forward headers X-FORWARDED-HOST and X-FORWARDED-PROTO.
     text: |
         az containerapp auth update -g myResourceGroup --name my-containerapp --proxy-convention Standard
+  - name: Configure the blob storage token store using system assigned managed identity to authenticate.
+    text: |
+        az containerapp auth update -g myResourceGroup --name my-containerapp --token-store true --blob-container-uri https://storageAccount1.blob.core.windows.net/container1
+  - name: Configure the blob storage token store using user assigned managed identity to authenticate.
+    text: |
+        az containerapp auth update -g myResourceGroup --name my-containerapp --token-store true --blob-container-uri https://storageAccount1.blob.core.windows.net/container1 --blob-container-identity managedIdentityResourceId
 """
 
 helps['containerapp env workload-profile set'] = """

src/containerapp/azext_containerapp/_params.py

@@ -276,6 +276,11 @@ def load_arguments(self, _):
         c.argument('user_assigned', nargs='+', help="Space-separated user identities to be assigned.")
         c.argument('system_assigned', help="Boolean indicating whether to assign system-assigned identity.", action='store_true')
 
+    with self.argument_context('containerapp auth') as c:
+        c.argument('blob_container_uri', help='The URI of the blob storage containing the tokens. Should not be used along with sas_url_secret and sas_url_secret_name.', is_preview=True)
+        c.argument('blob_container_identity', options_list=['--blob-container-identity', '--bci'],
+                   help='Resource ID of a managed identity to authenticate with Azure blob storage, or Empty to use a system-assigned identity.', is_preview=True)
+
     with self.argument_context('containerapp env workload-profile set') as c:
         c.argument('workload_profile_type', help="The type of workload profile to add or update. Run `az containerapp env workload-profile list-supported -l <region>` to check the options for your region.")
         c.argument('min_nodes', help="The minimum node count for the workload profile")

src/containerapp/azext_containerapp/containerapp_auth_decorator.py

@@ -3,11 +3,110 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 # pylint: disable=line-too-long, useless-parent-delegation
-
+from azure.cli.core.azclierror import ArgumentUsageError
+from azure.cli.core.commands.client_factory import get_subscription_id
 from azure.cli.command_modules.containerapp.containerapp_auth_decorator import ContainerAppAuthDecorator
-
+from azure.cli.command_modules.containerapp._utils import safe_set, set_field_in_auth_settings, update_http_settings_in_auth_settings, _ensure_identity_resource_id
+from azure.cli.command_modules.containerapp._constants import BLOB_STORAGE_TOKEN_STORE_SECRET_SETTING_NAME
 
 # decorator for preview auth show/update
 class ContainerAppPreviewAuthDecorator(ContainerAppAuthDecorator):
+
+    def parent_construct_payload(self):
+        self.existing_auth = {}
+        try:
+            self.existing_auth = self.client.get(cmd=self.cmd, resource_group_name=self.get_argument_resource_group_name(), container_app_name=self.get_argument_name(), auth_config_name="current")["properties"]
+        except:
+            self.existing_auth["platform"] = {}
+            self.existing_auth["platform"]["enabled"] = True
+            self.existing_auth["globalValidation"] = {}
+            self.existing_auth["login"] = {}
+
+        self.existing_auth = set_field_in_auth_settings(self.existing_auth, self.get_argument_set_string())
+
+        if self.get_argument_enabled() is not None:
+            if "platform" not in self.existing_auth:
+                self.existing_auth["platform"] = {}
+            self.existing_auth["platform"]["enabled"] = self.get_argument_enabled()
+
+        if self.get_argument_runtime_version() is not None:
+            if "platform" not in self.existing_auth:
+                self.existing_auth["platform"] = {}
+            self.existing_auth["platform"]["runtimeVersion"] = self.get_argument_runtime_version()
+
+        if self.get_argument_config_file_path() is not None:
+            if "platform" not in self.existing_auth:
+                self.existing_auth["platform"] = {}
+            self.existing_auth["platform"]["configFilePath"] = self.get_argument_config_file_path()
+
+        if self.get_argument_unauthenticated_client_action() is not None:
+            if "globalValidation" not in self.existing_auth:
+                self.existing_auth["globalValidation"] = {}
+            self.existing_auth["globalValidation"]["unauthenticatedClientAction"] = self.get_argument_unauthenticated_client_action()
+
+        if self.get_argument_redirect_provider() is not None:
+            if "globalValidation" not in self.existing_auth:
+                self.existing_auth["globalValidation"] = {}
+            self.existing_auth["globalValidation"]["redirectToProvider"] = self.get_argument_redirect_provider()
+
+        if self.get_argument_excluded_paths() is not None:
+            if "globalValidation" not in self.existing_auth:
+                self.existing_auth["globalValidation"] = {}
+            self.existing_auth["globalValidation"]["excludedPaths"] = self.get_argument_excluded_paths().split(",")
+
+        self.existing_auth = update_http_settings_in_auth_settings(self.existing_auth, self.get_argument_require_https(),
+                                                                   self.get_argument_proxy_convention(), self.get_argument_proxy_custom_host_header(),
+                                                                   self.get_argument_proxy_custom_proto_header())
+        
     def construct_payload(self):
-        super().construct_payload()
+        self.parent_construct_payload()
+        self.set_up_token_store()
+
+    def set_up_token_store(self):
+        if self.get_argument_token_store() is None:
+            return
+
+        if self.get_argument_token_store() is False:
+            safe_set(self.existing_auth, "login", "tokenStore", "enabled", value=False)
+            return
+
+        safe_set(self.existing_auth, "login", "tokenStore", "enabled", value=True)
+        safe_set(self.existing_auth, "login", "tokenStore", "azureBlobStorage", value={})
+
+        param_provided = 0
+        if self.get_argument_sas_url_secret() is not None:
+            param_provided += 1
+        if self.get_argument_sas_url_secret_name() is not None:
+            param_provided += 1
+        if self.get_argument_blob_container_uri() is not None:
+            param_provided += 1
+
+        if param_provided != 1:
+            raise ArgumentUsageError(
+                'Usage Error: only blob storage token store is supported. --sas-url-secret, --sas-url-secret-name and --blob-container-uri should provide exactly one when token store is enabled')
+        
+        if self.get_argument_blob_container_uri() is not None:
+            safe_set(self.existing_auth, "login", "tokenStore", "azureBlobStorage", "blobContainerUri",
+                    value=self.get_argument_blob_container_uri())
+
+            identity = self.get_argument_blob_container_identity()
+            if identity is None:
+                identity = ""
+            identity = identity.lower()
+            if identity != "":
+                subscription_id = get_subscription_id(self.cmd.cli_ctx)
+                identity = _ensure_identity_resource_id(subscription_id, self.get_argument_resource_group_name(), identity)
+                safe_set(self.existing_auth, "login", "tokenStore", "azureBlobStorage", "managedIdentityResourceId",
+                    value=identity)
+            return
+
+        sas_url_setting_name = BLOB_STORAGE_TOKEN_STORE_SECRET_SETTING_NAME
+        if self.get_argument_sas_url_secret_name() is not None:
+            sas_url_setting_name = self.get_argument_sas_url_secret_name()
+        safe_set(self.existing_auth, "login", "tokenStore", "azureBlobStorage", "sasUrlSettingName", value=sas_url_setting_name)
+    
+    def get_argument_blob_container_uri(self):
+        return self.get_param("blob_container_uri")
+
+    def get_argument_blob_container_identity(self):
+        return self.get_param("blob_container_identity")

src/containerapp/azext_containerapp/custom.py

@@ -1401,6 +1401,7 @@ def update_auth_config(cmd, resource_group_name, name, set_string=None, enabled=
                        proxy_convention=None, proxy_custom_host_header=None,
                        proxy_custom_proto_header=None, excluded_paths=None,
                        token_store=None, sas_url_secret=None, sas_url_secret_name=None,
+                       blob_container_uri=None, blob_container_identity=None,
                        yes=False):
     raw_parameters = locals()
     containerapp_auth_decorator = ContainerAppPreviewAuthDecorator(