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

scripts/ci/credscan/CredScanSuppressions.json

@@ -17,6 +17,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

@@ -11,6 +11,7 @@ Release History
 
 * Support import/export of keyvault references from/to appservice
 * Support import/export of all labels from appconfig to appconfig
+* Validate key and feature names before setting and importing
 
 **AppService**
 

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

@@ -3,10 +3,11 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
-# pylint: disable=line-too-long
+# pylint: disable=line-too-long,too-many-nested-blocks
 
 import io
 import json
+import re
 import sys
 
 import chardet
@@ -88,6 +89,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(FeatureFlagConstants.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
 
 
@@ -137,7 +168,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
 
 
@@ -273,45 +305,46 @@ 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']
-            tags = {'AppService:SlotSetting': str(item['slotSetting']).lower()} if item['slotSetting'] else {}
-            value = item['value']
-
-            # Value will look like one of the following if it is a KeyVault reference:
-            # @Microsoft.KeyVault(SecretUri=https://myvault.vault.azure.net/secrets/mysecret/ec96f02080254f109c51a1f14cdb1931)
-            # @Microsoft.KeyVault(VaultName=myvault;SecretName=mysecret;SecretVersion=ec96f02080254f109c51a1f14cdb1931)
-            if value and value.strip().lower().startswith(KeyVaultConstants.APPSVC_KEYVAULT_PREFIX.lower()):
-                try:
-                    # Strip all whitespaces from value string.
-                    # Valid values of SecretUri, VaultName, SecretName or SecretVersion will never have whitespaces.
-                    value = value.replace(" ", "")
-                    appsvc_value_dict = dict(x.split('=') for x in value[len(KeyVaultConstants.APPSVC_KEYVAULT_PREFIX) + 1: -1].split(';'))
-                    appsvc_value_dict = {k.lower(): v for k, v in appsvc_value_dict.items()}
-                    secret_identifier = appsvc_value_dict.get('secreturi')
-                    if not secret_identifier:
-                        # Construct secreturi
-                        vault_name = appsvc_value_dict.get('vaultname')
-                        secret_name = appsvc_value_dict.get('secretname')
-                        secret_version = appsvc_value_dict.get('secretversion')
-                        secret_identifier = "https://{0}.vault.azure.net/secrets/{1}/{2}".format(vault_name, secret_name, secret_version)
+            if validate_import_key(key):
+                tags = {'AppService:SlotSetting': str(item['slotSetting']).lower()} if item['slotSetting'] else {}
+                value = item['value']
+
+                # Value will look like one of the following if it is a KeyVault reference:
+                # @Microsoft.KeyVault(SecretUri=https://myvault.vault.azure.net/secrets/mysecret/ec96f02080254f109c51a1f14cdb1931)
+                # @Microsoft.KeyVault(VaultName=myvault;SecretName=mysecret;SecretVersion=ec96f02080254f109c51a1f14cdb1931)
+                if value and value.strip().lower().startswith(KeyVaultConstants.APPSVC_KEYVAULT_PREFIX.lower()):
                     try:
-                        from azure.keyvault.key_vault_id import KeyVaultIdentifier
-                        # this throws an exception for invalid format of secret identifier
-                        KeyVaultIdentifier(uri=secret_identifier)
-                        kv = KeyValue(key=key,
-                                      value=json.dumps({"uri": secret_identifier}, ensure_ascii=False, separators=(',', ':')),
-                                      tags=tags,
-                                      content_type=KeyVaultConstants.KEYVAULT_CONTENT_TYPE)
-                        key_values.append(kv)
-                        continue
-                    except (TypeError, ValueError) as e:
+                        # Strip all whitespaces from value string.
+                        # Valid values of SecretUri, VaultName, SecretName or SecretVersion will never have whitespaces.
+                        value = value.replace(" ", "")
+                        appsvc_value_dict = dict(x.split('=') for x in value[len(KeyVaultConstants.APPSVC_KEYVAULT_PREFIX) + 1: -1].split(';'))
+                        appsvc_value_dict = {k.lower(): v for k, v in appsvc_value_dict.items()}
+                        secret_identifier = appsvc_value_dict.get('secreturi')
+                        if not secret_identifier:
+                            # Construct secreturi
+                            vault_name = appsvc_value_dict.get('vaultname')
+                            secret_name = appsvc_value_dict.get('secretname')
+                            secret_version = appsvc_value_dict.get('secretversion')
+                            secret_identifier = "https://{0}.vault.azure.net/secrets/{1}/{2}".format(vault_name, secret_name, secret_version)
+                        try:
+                            from azure.keyvault.key_vault_id import KeyVaultIdentifier
+                            # this throws an exception for invalid format of secret identifier
+                            KeyVaultIdentifier(uri=secret_identifier)
+                            kv = KeyValue(key=key,
+                                          value=json.dumps({"uri": secret_identifier}, ensure_ascii=False, separators=(',', ':')),
+                                          tags=tags,
+                                          content_type=KeyVaultConstants.KEYVAULT_CONTENT_TYPE)
+                            key_values.append(kv)
+                            continue
+                        except (TypeError, ValueError) as e:
+                            logger.debug(
+                                'Exception while validating the format of KeyVault identifier. Key "%s" with value "%s" will be treated like a regular key-value.\n%s', key, value, str(e))
+                    except (AttributeError, TypeError, ValueError) as e:
                         logger.debug(
-                            'Exception while validating the format of KeyVault identifier. Key "%s" with value "%s" will be treated like a regular key-value.\n%s', key, value, str(e))
-                except (AttributeError, TypeError, ValueError) as e:
-                    logger.debug(
-                        'Key "%s" with value "%s" is not a well-formatted KeyVault reference. It will be treated like a regular key-value.\n%s', key, value, str(e))
+                            'Key "%s" with value "%s" is not a well-formatted KeyVault reference. It will be treated like a regular key-value.\n%s', key, value, str(e))
 
-            kv = KeyValue(key=key, value=value, tags=tags)
-            key_values.append(kv)
+                kv = KeyValue(key=key, value=value, tags=tags)
+                key_values.append(kv)
         return key_values
     except Exception as exception:
         raise CLIError("Failed to read key-values from appservice." + str(exception))
@@ -699,47 +732,49 @@ def __convert_feature_dict_to_keyvalue_list(features_dict, enabled_for_keyword):
 
     try:
         for k, v in features_dict.items():
-            key = FeatureFlagConstants.FEATURE_FLAG_PREFIX + str(k)
-            feature_flag_value = FeatureFlagValue(id_=str(k))
+            if validate_import_feature(feature=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
+                key = FeatureFlagConstants.FEATURE_FLAG_PREFIX + str(k)
+                feature_flag_value = FeatureFlagValue(id_=str(k))
 
-                        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 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=FeatureFlagConstants.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=FeatureFlagConstants.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, _):
@@ -124,14 +125,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.")
@@ -170,7 +171,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):
@@ -162,3 +165,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_features_prop.json

@@ -1,4 +1,5 @@
 #Mon Jan 20 09:34:31 Pacific Standard Time 2020
+
 Color=Red
 Region=West US
 feature-management.FalseFeature=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
+  }
+}