[AppConfig] Validate key and feature names before setting and importing

CredScanSuppressions.json

@@ -16,6 +16,7 @@
         "src\\azure-cli\\azure\\cli\\command_modules\\appconfig\\tests\\latest\\recordings\\test_azconfig_feature_filter.yaml",
         "src\\azure-cli\\azure\\cli\\command_modules\\appconfig\\tests\\latest\\recordings\\test_azconfig_import_export_naming_conventions.yaml",
         "src\\azure-cli\\azure\\cli\\command_modules\\appconfig\\tests\\latest\\recordings\\test_azconfig_import_export.yaml",
+        "src\\azure-cli\\azure\\cli\\command_modules\\appconfig\\tests\\latest\\recordings\\test_azconfig_key_validation.yaml",
         "src\\azure-cli\\azure\\cli\\command_modules\\appconfig\\tests\\latest\\recordings\\test_azconfig_kv.yaml"
       ],
       "_justification": "[AppConfig] response body contains random value recognized as secret"

src/azure-cli/HISTORY.rst

@@ -3,6 +3,10 @@
 Release History
 ===============
 
+**AppConfig**
+
+* Validate key and feature names before setting and importing
+
 2.0.79
 ++++++
 

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

@@ -7,6 +7,7 @@
 
 import io
 import json
+import re
 import sys
 
 import chardet
@@ -89,6 +90,36 @@ def __compare_kvs_for_restore(restore_kvs, current_kvs):
 
     return kvs_to_restore, kvs_to_modify, kvs_to_delete
 
+
+def validate_import_key(key):
+    if key:
+        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
+        if key.startswith(FEATURE_FLAG_PREFIX):
+            logger.warning("Ignoring invalid key '%s'. Key cannot start with the reserved prefix for feature flags.", key)
+            return False
+    else:
+        logger.warning("Ignoring invalid key ''. Key cannot be empty.")
+        return False
+
+    return True
+
+
+def validate_import_feature(feature):
+    if feature:
+        invalid_pattern = re.compile(r'[^a-zA-Z0-9._-]')
+        invalid = re.search(invalid_pattern, feature)
+        if invalid:
+            logger.warning("Ignoring invalid feature '%s'. Only alphanumeric characters, '.', '-' and '_' are allowed in feature name.", feature)
+            return False
+    else:
+        logger.warning("Ignoring invalid feature ''. Feature name cannot be empty.")
+        return False
+
+    return True
+
+
 # File <-> List of KeyValue object
 
 
@@ -138,7 +169,8 @@ def __read_kv_from_file(file_path, format_, separator=None, prefix_to_add="", de
     # convert to KeyValue list
     key_values = []
     for k, v in flattened_data.items():
-        key_values.append(KeyValue(key=k, value=v))
+        if validate_import_key(key=k):
+            key_values.append(KeyValue(key=k, value=v))
     return key_values
 
 
@@ -269,14 +301,14 @@ def __read_kv_from_app_service(cmd, appservice_account, prefix_to_add=""):
             cmd, resource_group_name=appservice_account["resource_group"], name=appservice_account["name"], slot=None)
         for item in settings:
             key = prefix_to_add + item['name']
-            value = item['value']
-            tags = {'AppService:SlotSetting': str(item['slotSetting']).lower()} if item['slotSetting'] else {}
-            kv = KeyValue(key=key, value=value, tags=tags)
-            key_values.append(kv)
+            if validate_import_key(key):
+                value = item['value']
+                tags = {'AppService:SlotSetting': str(item['slotSetting']).lower()} if item['slotSetting'] else {}
+                kv = KeyValue(key=key, value=value, tags=tags)
+                key_values.append(kv)
         return key_values
     except Exception as exception:
-        raise CLIError(
-            "Fail to read key-values from appservice." + str(exception))
+        raise CLIError("Fail to read key-values from appservice." + str(exception))
 
 
 def __write_kv_to_app_service(cmd, key_values, appservice_account):
@@ -295,8 +327,7 @@ def __write_kv_to_app_service(cmd, key_values, appservice_account):
         update_app_settings(cmd, resource_group_name=appservice_account["resource_group"],
                             name=appservice_account["name"], settings=non_slot_settings, slot_settings=slot_settings)
     except Exception as exception:
-        raise CLIError(
-            "Fail to write key-values to appservice: " + str(exception))
+        raise CLIError("Fail to write key-values to appservice: " + str(exception))
 
 
 # Helper functions
@@ -649,47 +680,48 @@ def __convert_feature_dict_to_keyvalue_list(features_dict, enabled_for_keyword):
 
     try:
         for k, v in features_dict.items():
-            key = FEATURE_FLAG_PREFIX + str(k)
-            feature_flag_value = FeatureFlagValue(id_=str(k))
-
-            if isinstance(v, dict):
-                # This may be a conditional feature
-                feature_flag_value.enabled = False
-                try:
-                    feature_flag_value.conditions = {'client_filters': v[enabled_for_keyword]}
-                except KeyError:
-                    raise CLIError("Feature '{0}' must contain '{1}' definition or have a true/false value. \n".format(str(k), enabled_for_keyword))
-
-                if feature_flag_value.conditions["client_filters"]:
-                    feature_flag_value.enabled = True
-
-                    for idx, val in enumerate(feature_flag_value.conditions["client_filters"]):
-                        # each val should be a dict with at most 2 keys (Name, Parameters) or at least 1 key (Name)
-                        val = {filter_key.lower(): filter_val for filter_key, filter_val in val.items()}
-                        if not val.get("name", None):
-                            logger.warning("Ignoring a filter for feature '%s' because it doesn't have a 'Name' attribute.", str(k))
-                            continue
-
-                        if val["name"].lower() == "alwayson":
-                            # We support alternate format for specifying always ON features
-                            # "FeatureT": {"EnabledFor": [{ "Name": "AlwaysOn"}]}
-                            feature_flag_value.conditions = default_conditions
-                            break
-
-                        filter_param = val.get("parameters", {})
-                        new_val = {'name': val["name"]}
-                        if filter_param:
-                            new_val["parameters"] = filter_param
-                        feature_flag_value.conditions["client_filters"][idx] = new_val
+            if validate_import_feature(feature=k):
+                key = FEATURE_FLAG_PREFIX + str(k)
+                feature_flag_value = FeatureFlagValue(id_=str(k))
+
+                if isinstance(v, dict):
+                    # This may be a conditional feature
+                    feature_flag_value.enabled = False
+                    try:
+                        feature_flag_value.conditions = {'client_filters': v[enabled_for_keyword]}
+                    except KeyError:
+                        raise CLIError("Feature '{0}' must contain '{1}' definition or have a true/false value. \n".format(str(k), enabled_for_keyword))
+
+                    if feature_flag_value.conditions["client_filters"]:
+                        feature_flag_value.enabled = True
+
+                        for idx, val in enumerate(feature_flag_value.conditions["client_filters"]):
+                            # each val should be a dict with at most 2 keys (Name, Parameters) or at least 1 key (Name)
+                            val = {filter_key.lower(): filter_val for filter_key, filter_val in val.items()}
+                            if not val.get("name", None):
+                                logger.warning("Ignoring a filter for feature '%s' because it doesn't have a 'Name' attribute.", str(k))
+                                continue
+
+                            if val["name"].lower() == "alwayson":
+                                # We support alternate format for specifying always ON features
+                                # "FeatureT": {"EnabledFor": [{ "Name": "AlwaysOn"}]}
+                                feature_flag_value.conditions = default_conditions
+                                break
+
+                            filter_param = val.get("parameters", {})
+                            new_val = {'name': val["name"]}
+                            if filter_param:
+                                new_val["parameters"] = filter_param
+                            feature_flag_value.conditions["client_filters"][idx] = new_val
 
-            else:
-                feature_flag_value.enabled = v
-                feature_flag_value.conditions = default_conditions
+                else:
+                    feature_flag_value.enabled = v
+                    feature_flag_value.conditions = default_conditions
 
-            set_kv = KeyValue(key=key,
-                              value=json.dumps(feature_flag_value, default=lambda o: o.__dict__, ensure_ascii=False),
-                              content_type=FEATURE_FLAG_CONTENT_TYPE)
-            key_values.append(set_kv)
+                set_kv = KeyValue(key=key,
+                                  value=json.dumps(feature_flag_value, default=lambda o: o.__dict__, ensure_ascii=False),
+                                  content_type=FEATURE_FLAG_CONTENT_TYPE)
+                key_values.append(set_kv)
 
     except Exception as exception:
         raise CLIError("File contains feature flags in invalid format. " + str(exception))

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

@@ -20,7 +20,8 @@
                           validate_export, validate_import,
                           validate_import_depth, validate_query_fields,
                           validate_feature_query_fields, validate_filter_parameters,
-                          validate_separator, validate_secret_identifier)
+                          validate_separator, validate_secret_identifier,
+                          validate_key, validate_content_type, validate_feature)
 
 
 def load_arguments(self, _):
@@ -122,14 +123,14 @@ def load_arguments(self, _):
         c.argument('appservice_account', validator=validate_appservice_name_or_id, help='ARM ID for AppService OR the name of the AppService, assuming it is in the same subscription and resource group as the App Configuration. Required for AppService arguments')
 
     with self.argument_context('appconfig kv set') as c:
-        c.argument('key', help='Key to be set.')
+        c.argument('key', validator=validate_key, help="Key to be set. Key cannot be a '.' or '..', or contain the '%' character.")
         c.argument('label', help="If no label specified, set the key with null label by default")
         c.argument('tags', arg_type=tags_type)
-        c.argument('content_type', help='Content type of the keyvalue to be set.')
+        c.argument('content_type', validator=validate_content_type, help='Content type of the keyvalue to be set.')
         c.argument('value', help='Value of the keyvalue to be set.')
 
     with self.argument_context('appconfig kv set-keyvault') as c:
-        c.argument('key', help='Key to be set.')
+        c.argument('key', validator=validate_key, help="Key to be set. Key cannot be a '.' or '..', or contain the '%' character.")
         c.argument('label', help="If no label specified, set the key with null label by default")
         c.argument('tags', arg_type=tags_type)
         c.argument('secret_identifier', validator=validate_secret_identifier, help="ID of the Key Vault object. Can be found using 'az keyvault {collection} show' command, where collection is key, secret or certificate. To set reference to the latest version of your secret, remove version information from secret identifier.")
@@ -168,7 +169,7 @@ def load_arguments(self, _):
         c.argument('fields', arg_type=feature_fields_arg_type)
 
     with self.argument_context('appconfig feature set') as c:
-        c.argument('feature', help="Name of the feature flag to be set. Only alphanumeric characters, '.', '-' and '_' are allowed.")
+        c.argument('feature', validator=validate_feature, help="Name of the feature flag to be set. Only alphanumeric characters, '.', '-' and '_' are allowed.")
         c.argument('label', help="If no label specified, set the feature flag with null label by default")
         c.argument('description', help='Description of the feature flag to be set.')
 

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

@@ -14,6 +14,9 @@
 from ._featuremodels import FeatureQueryFields
 
 logger = get_logger(__name__)
+FEATURE_FLAG_PREFIX = ".appconfig.featureflag/"
+FEATURE_FLAG_CONTENT_TYPE = "application/vnd.microsoft.appconfig.ff+json;charset=utf-8"
+KEYVAULT_CONTENT_TYPE = "application/vnd.microsoft.appconfig.keyvaultref+json;charset=utf-8"
 
 
 def validate_datetime(namespace):
@@ -159,3 +162,33 @@ def validate_secret_identifier(namespace):
         KeyVaultIdentifier(uri=identifier)
     except Exception as e:
         raise CLIError("Received an exception while validating the format of secret identifier.\n{0}".format(str(e)))
+
+
+def validate_key(namespace):
+    if namespace.key:
+        input_key = str(namespace.key).lower()
+        if input_key == '.' or input_key == '..' or '%' in input_key:
+            raise CLIError("Key is invalid. Key cannot be a '.' or '..', or contain the '%' character.")
+        if input_key.startswith(FEATURE_FLAG_PREFIX):
+            raise CLIError("Key is invalid. Key cannot start with the reserved prefix for feature flags.")
+    else:
+        raise CLIError("Key cannot be empty.")
+
+
+def validate_content_type(namespace):
+    if namespace.content_type is not None:
+        content_type = str(namespace.content_type).lower()
+        if content_type == FEATURE_FLAG_CONTENT_TYPE:
+            raise CLIError("Content type is invalid. It's a reserved content type for feature flags.")
+        if content_type == KEYVAULT_CONTENT_TYPE:
+            raise CLIError("Content type is invalid. It's a reserved content type for KeyVault references.")
+
+
+def validate_feature(namespace):
+    if namespace.feature:
+        invalid_pattern = re.compile(r'[^a-zA-Z0-9._-]')
+        invalid = re.search(invalid_pattern, namespace.feature)
+        if invalid:
+            raise CLIError("Feature name is invalid. Only alphanumeric characters, '.', '-' and '_' are allowed.")
+    else:
+        raise CLIError("Feature name cannot be empty.")

src/azure-cli/azure/cli/command_modules/appconfig/tests/latest/export.json

@@ -1,20 +1,6 @@
 {
-  "BackgroundColor": "red",
   "Language": "spanish",
-  "Langugage": "English",
-  "Settings:BackgroundColor": {
-    "cars": {
-      "Toyota": "Acura"
-    },
-    "ship": [
-      "ship1",
-      "ship2",
-      "ship3"
-    ]
-  },
-  "abc": "",
-  "appabc": "",
-  "background-color": "black",
-  "font-size": "34",
-  "language": "spanish"
+  "FeatureManagement": {
+    "Beta": false
+  }
 }

src/azure-cli/azure/cli/command_modules/appconfig/tests/latest/export_valid_kv_and_features.json

@@ -0,0 +1,6 @@
+{
+  "Language": "spanish",
+  "FeatureManagement": {
+    "Beta": false
+  }
+}

src/azure-cli/azure/cli/command_modules/appconfig/tests/latest/import_invalid_kv_and_features.json

@@ -0,0 +1,8 @@
+{
+  "Background%%Color": "red",
+  "Language": "spanish",
+  "FeatureManagement": {
+    "Beta": false,
+    "$Percentage$": true
+  }
+}