{App Config} Fix #12668860, #12668922: Fix validations while importing kvset

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

@@ -148,7 +148,7 @@
   - name: Import all keys to another App Configuration using your 'az login' credentials.
     text: az appconfig kv import -s appconfig --endpoint https://myappconfiguration.azconfig.io --auth-mode login --src-endpoint https://anotherappconfiguration.azconfig.io --src-auth-mode login --src-key * --src-label * --preserve-labels
   - name: Import all keys and feature flags from a file using the appconfig/kvset format.
-    text: az appconfig kv import -n MyAppConfiguration --label test -s file --path D:/abc.json --format json --profile appconfig/kvset
+    text: az appconfig kv import -n MyAppConfiguration -s file --path D:/abc.json --format json --profile appconfig/kvset
 """
 
 helps['appconfig kv list'] = """

src/azure-cli/azure/cli/command_modules/appconfig/_kv_helpers.py

@@ -3,7 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
-# pylint: disable=line-too-long,too-many-nested-blocks,too-many-lines
+# pylint: disable=line-too-long,too-many-nested-blocks,too-many-lines,too-many-return-statements
 
 import io
 import json
@@ -97,6 +97,9 @@ def __compare_kvs_for_restore(restore_kvs, current_kvs):
 
 def validate_import_key(key):
     if key:
+        if not isinstance(key, str):
+            logger.warning("Ignoring invalid key '%s'. Key must be a string.", key)
+            return False
         if key == '.' or key == '..' or '%' in key:
             logger.warning("Ignoring invalid key '%s'. Key cannot be a '.' or '..', or contain the '%%' character.", key)
             return False
@@ -106,7 +109,6 @@ def validate_import_key(key):
     else:
         logger.warning("Ignoring invalid key ''. Key cannot be empty.")
         return False
-
     return True
 
 
@@ -380,13 +382,13 @@ def __write_kv_and_features_to_config_store(azconfig_client,
 
 
 def __is_feature_flag(kv):
-    if kv and kv.key and kv.content_type:
+    if kv and kv.key and isinstance(kv.key, str) and kv.content_type and isinstance(kv.content_type, str):
         return kv.key.startswith(FeatureFlagConstants.FEATURE_FLAG_PREFIX) and kv.content_type == FeatureFlagConstants.FEATURE_FLAG_CONTENT_TYPE
     return False
 
 
 def __is_key_vault_ref(kv):
-    return kv and kv.content_type and kv.content_type.lower() == KeyVaultConstants.KEYVAULT_CONTENT_TYPE
+    return kv and kv.content_type and isinstance(kv.content_type, str) and kv.content_type.lower() == KeyVaultConstants.KEYVAULT_CONTENT_TYPE
 
 
 def __discard_features_from_retrieved_kv(src_kvs):
@@ -1027,13 +1029,19 @@ def __import_kvset_from_file(client, path, yes):
     if KVSetConstants.KVSETRootElementName not in new_kvset:
         raise FileOperationError("file '{0}' is not in a valid '{1}' format.".format(path, ImportExportProfiles.KVSET))
 
-    kvset_to_import = [ConfigurationSetting(key=kv['key'],
-                                            label=kv['label'],
-                                            content_type=kv['content_type'],
-                                            value=kv['value'],
-                                            tags=kv['tags'])
+    kvset_from_file = [ConfigurationSetting(key=kv.get('key', None),
+                                            label=kv.get('label', None),
+                                            content_type=kv.get('content_type', None),
+                                            value=kv.get('value', None),
+                                            tags=kv.get('tags', None))
                        for kv in new_kvset[KVSetConstants.KVSETRootElementName]]
 
+    kvset_to_import = []
+
+    for config_setting in kvset_from_file:
+        if __validate_import_config_setting(config_setting):
+            kvset_to_import.append(config_setting)
+
     if not yes:
         existing_kvset = __read_kv_from_config_store(client,
                                                      key=SearchFilterOptions.ANY_KEY,
@@ -1056,16 +1064,6 @@ def __import_kvset_from_file(client, path, yes):
         user_confirmation('Do you want to continue?\n')
 
     for config_setting in kvset_to_import:
-        if __is_key_vault_ref(kv=config_setting):
-            if not __validate_import_keyvault_ref(kv=config_setting):
-                continue
-        elif __is_feature_flag(kv=config_setting):
-            if not __validate_import_feature_flag(kv=config_setting):
-                continue
-        elif not validate_import_key(config_setting.key):
-            continue
-
-        # All validations successful
         __write_configuration_setting_to_config_store(client, config_setting)
 
 
@@ -1103,6 +1101,42 @@ def __validate_import_feature_flag(kv):
     return False
 
 
+def __validate_import_config_setting(config_setting):
+    if __is_key_vault_ref(kv=config_setting):
+        if not __validate_import_keyvault_ref(kv=config_setting):
+            return False
+    elif __is_feature_flag(kv=config_setting):
+        if not __validate_import_feature_flag(kv=config_setting):
+            return False
+    elif not validate_import_key(config_setting.key):
+        return False
+
+    if config_setting.value and not isinstance(config_setting.value, str):
+        logger.warning("The 'value' for the key '{%s}' is not a string. This key-value will not be imported.", config_setting.key)
+        return False
+    if config_setting.content_type and not isinstance(config_setting.content_type, str):
+        logger.warning("The 'content_type' for the key '{%s}' is not a string. This key-value will not be imported.", config_setting.key)
+        return False
+    if config_setting.label and not isinstance(config_setting.label, str):
+        logger.warning("The 'label' for the key '{%s}' is not a string. This key-value will not be imported.", config_setting.key)
+        return False
+
+    return __validate_import_tags(config_setting)
+
+
+def __validate_import_tags(kv):
+    if kv.tags and not isinstance(kv.tags, dict):
+        logger.warning("The format of 'tags' for key '%s' is not valid. This key-value will not be imported.", kv.key)
+        return False
+
+    if kv.tags:
+        for tag_key, tag_value in kv.tags.items():
+            if not isinstance(tag_value, str):
+                logger.warning("The value for the tag '{%s}' for key '{%s}' is not in a valid format. This key-value will not be imported.", tag_key, kv.key)
+                return False
+    return True
+
+
 def __write_configuration_setting_to_config_store(azconfig_client, configuration_setting):
     try:
         azconfig_client.set_configuration_setting(configuration_setting)