Joinnis/add validators

src/k8sconfiguration/azext_k8sconfiguration/_params.py

@@ -13,26 +13,36 @@
 )
 
 from azure.cli.core.commands.validators import get_default_location_from_resource_group
-from ._validators import validate_configuration_type
+from ._validators import validate_configuration_type, validate_configuration_name, validate_operator_namespace, validate_operator_instance_name
 
 
 def load_arguments(self, _):
     sourcecontrolconfiguration_type = CLIArgumentType(help='Name of the Kubernetes Configuration')
 
     with self.argument_context('k8sconfiguration') as c:
         c.argument('tags', tags_type)
-        c.argument('location', validator=get_default_location_from_resource_group)
-        c.argument('name', sourcecontrolconfiguration_type, options_list=['--name', '-n'])
-        c.argument('cluster_name', options_list=['--cluster-name', '-c'], help='Name of the Kubernetes cluster')
-        c.argument('cluster_type', arg_type=get_enum_type(['connectedClusters', 'managedClusters']),
+        c.argument('location',
+                   validator=get_default_location_from_resource_group)
+        c.argument('name', sourcecontrolconfiguration_type,
+                   options_list=['--name', '-n'],
+                   validator=validate_configuration_name)
+        c.argument('cluster_name',
+                   options_list=['--cluster-name', '-c'],
+                   help='Name of the Kubernetes cluster')
+        c.argument('cluster_type',
+                   arg_type=get_enum_type(['connectedClusters', 'managedClusters']),
                    help='Specify Arc clusters or AKS managed clusters.')
-        c.argument('repository_url', options_list=['--repository-url', '-u'],
+        c.argument('repository_url',
+                   options_list=['--repository-url', '-u'],
                    help='Url of the source control repository')
-        c.argument('enable_helm_operator', arg_type=get_three_state_flag(),
+        c.argument('enable_helm_operator',
+                   arg_type=get_three_state_flag(),
                    help='Enable support for Helm chart deployments')
-        c.argument('scope', arg_type=get_enum_type(['namespace', 'cluster']),
+        c.argument('scope',
+                   arg_type=get_enum_type(['namespace', 'cluster']),
                    help='''Specify scope of the operator to be 'namespace' or 'cluster' ''')
-        c.argument('configuration_type', validator=validate_configuration_type,
+        c.argument('configuration_type',
+                   validator=validate_configuration_type,
                    arg_type=get_enum_type(['sourceControlConfiguration']),
                    help='Type of the configuration')
         c.argument('helm_operator_params',
@@ -54,11 +64,13 @@ def load_arguments(self, _):
         c.argument('ssh_known_hosts_file',
                    help='Specify filepath to known_hosts contents containing public SSH keys required to access private Git instances')
         c.argument('operator_instance_name',
-                   help='Instance name of the Operator')
+                   help='Instance name of the Operator',
+                   validator=validate_operator_instance_name)
         c.argument('operator_namespace',
-                   help='Namespace in which to install the Operator')
+                   help='Namespace in which to install the Operator', 
+                   validator=validate_operator_namespace)
         c.argument('operator_type',
                    help='''Type of the operator. Valid value is 'flux' ''')
 
     with self.argument_context('k8sconfiguration list') as c:
-        c.argument('sourcecontrolconfiguration', sourcecontrolconfiguration_type, id_part=None)
+        c.argument('sourcecontrolconfiguration', sourcecontrolconfiguration_type, id_part=None)

src/k8sconfiguration/azext_k8sconfiguration/_validators.py

@@ -3,6 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
+import re
 from azure.cli.core.azclierror import InvalidArgumentValueError
 
 
@@ -11,3 +12,25 @@ def validate_configuration_type(configuration_type):
         raise InvalidArgumentValueError(
             'Invalid configuration-type',
             'Try specifying the valid value "sourceControlConfiguration"')
+
+def validate_configuration_name(namespace):
+    if namespace.name:
+        __validate_k8s_name(namespace.name, "configuration name", 63)
+
+def validate_operator_namespace(namespace):
+    if namespace.operator_namespace:
+        __validate_k8s_name(namespace.operator_namespace, "operator namespace", 23)
+
+def validate_operator_instance_name(namespace):
+    if namespace.operator_instance_name:
+        __validate_k8s_name(namespace.operator_instance_name, "operator instance name", 23)
+
+def __validate_k8s_name(param_value, param_name, max_len):
+    if len(param_value) > max_len:
+        raise InvalidArgumentValueError(
+            'Invalid {0} parameter. Valid {0}s can be a maximum of {1} characters'.format(param_name, max_len),
+            'Try shortening the length of your {0}'.format(param_name))
+    if not re.match(r'^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$', param_value):
+        raise InvalidArgumentValueError(
+            'Invalid {0} parameter. Valid {0}s can only contain lowercase alphanumeric characters and hyphens'.format(param_name),
+            'Try checking that your {0} only contains these characters'.format(param_name))

src/k8sconfiguration/azext_k8sconfiguration/custom.py

@@ -77,16 +77,16 @@ def create_k8sconfiguration(client, resource_group_name, cluster_name, name, rep
         helm_operator_properties.chart_version = helm_operator_version.strip()
         helm_operator_properties.chart_values = helm_operator_params.strip()
 
-    protected_settings = __get_protected_settings(ssh_private_key, ssh_private_key_file, https_user, https_key)
-    knownhost_data = __get_data_from_key_or_file(ssh_known_hosts, ssh_known_hosts_file)
+    protected_settings = get_protected_settings(ssh_private_key, ssh_private_key_file, https_user, https_key)
+    knownhost_data = get_data_from_key_or_file(ssh_known_hosts, ssh_known_hosts_file)
     if knownhost_data != '':
-        __validate_known_hosts(knownhost_data)
+        validate_known_hosts(knownhost_data)
 
     # Flag which parameters have been set and validate these settings against the set repository url
     ssh_private_key_set = ssh_private_key != '' or ssh_private_key_file != ''
     ssh_known_hosts_set = knownhost_data != ''
     https_auth_set = https_user != '' and https_key != ''
-    __validate_url_with_params(repository_url, ssh_private_key_set, ssh_known_hosts_set, https_auth_set)
+    validate_url_with_params(repository_url, ssh_private_key_set, ssh_known_hosts_set, https_auth_set)
 
     # Create sourceControlConfiguration object
     source_control_configuration = SourceControlConfiguration(repository_url=repository_url,
@@ -133,9 +133,9 @@ def update_k8sconfiguration(client, resource_group_name, cluster_name, name, clu
         config['operator_params'] = operator_params
         update_yes = True
 
-    knownhost_data = __get_data_from_key_or_file(ssh_known_hosts, ssh_known_hosts_file)
+    knownhost_data = get_data_from_key_or_file(ssh_known_hosts, ssh_known_hosts_file)
     if knownhost_data != '':
-        __validate_known_hosts(knownhost_data)
+        validate_known_hosts(knownhost_data)
         config['ssh_known_hosts_contents'] = knownhost_data
         update_yes = True
 
@@ -158,7 +158,7 @@ def update_k8sconfiguration(client, resource_group_name, cluster_name, name, clu
 
     # Flag which parameters have been set and validate these settings against the set repository url
     ssh_known_hosts_set = 'ssh_known_hosts_contents' in config
-    __validate_url_with_params(config['repository_url'], False, ssh_known_hosts_set, False)
+    validate_url_with_params(config['repository_url'], False, ssh_known_hosts_set, False)
 
     config = client.create_or_update(resource_group_name, cluster_rp, cluster_type, cluster_name,
                                      source_control_configuration_name, config)
@@ -183,28 +183,28 @@ def delete_k8sconfiguration(client, resource_group_name, cluster_name, name, clu
     return client.delete(resource_group_name, cluster_rp, cluster_type, cluster_name, source_control_configuration_name)
 
 
-def __get_protected_settings(ssh_private_key, ssh_private_key_file, https_user, https_key):
+def get_protected_settings(ssh_private_key, ssh_private_key_file, https_user, https_key):
     protected_settings = {}
-    ssh_private_key_data = __get_data_from_key_or_file(ssh_private_key, ssh_private_key_file)
+    ssh_private_key_data = get_data_from_key_or_file(ssh_private_key, ssh_private_key_file)
 
     # Add gitops private key data to protected settings if exists
     # Dry-run all key types to determine if the private key is in a valid format
     invalid_rsa_key, invalid_ecc_key, invalid_dsa_key, invalid_ed25519_key = (False, False, False, False)
     if ssh_private_key_data != '':
         try:
-            RSA.import_key(__from_base64(ssh_private_key_data))
+            RSA.import_key(from_base64(ssh_private_key_data))
         except ValueError:
             invalid_rsa_key = True
         try:
-            ECC.import_key(__from_base64(ssh_private_key_data))
+            ECC.import_key(from_base64(ssh_private_key_data))
         except ValueError:
             invalid_ecc_key = True
         try:
-            DSA.import_key(__from_base64(ssh_private_key_data))
+            DSA.import_key(from_base64(ssh_private_key_data))
         except ValueError:
             invalid_dsa_key = True
         try:
-            key_obj = io.StringIO(__from_base64(ssh_private_key_data).decode('utf-8'))
+            key_obj = io.StringIO(from_base64(ssh_private_key_data).decode('utf-8'))
             Ed25519Key(file_obj=key_obj)
         except SSHException:
             invalid_ed25519_key = True
@@ -217,8 +217,8 @@ def __get_protected_settings(ssh_private_key, ssh_private_key_file, https_user,
 
     # Check if both httpsUser and httpsKey exist, then add to protected settings
     if https_user != '' and https_key != '':
-        protected_settings['httpsUser'] = __to_base64(https_user)
-        protected_settings['httpsKey'] = __to_base64(https_key)
+        protected_settings['httpsUser'] = to_base64(https_user)
+        protected_settings['httpsKey'] = to_base64(https_key)
     elif https_user != '':
         raise RequiredArgumentMissingError(
             'Error! --https-user used without --https-key',
@@ -248,7 +248,7 @@ def __fix_compliance_state(config):
     return config
 
 
-def __validate_url_with_params(repository_url, ssh_private_key_set, known_hosts_contents_set, https_auth_set):
+def validate_url_with_params(repository_url, ssh_private_key_set, known_hosts_contents_set, https_auth_set):
     scheme = urlparse(repository_url).scheme
 
     if scheme in ('http', 'https'):
@@ -270,9 +270,9 @@ def __validate_url_with_params(repository_url, ssh_private_key_set, known_hosts_
                 'Verify the url provided is a valid http(s) url and not an ssh url')
 
 
-def __validate_known_hosts(knownhost_data):
+def validate_known_hosts(knownhost_data):
     try:
-        knownhost_str = __from_base64(knownhost_data).decode('utf-8')
+        knownhost_str = from_base64(knownhost_data).decode('utf-8')
     except Exception as ex:
         raise InvalidArgumentValueError(
             'Error! ssh known_hosts is not a valid utf-8 base64 encoded string',
@@ -293,38 +293,38 @@ def __validate_known_hosts(knownhost_data):
                 'Verify that all lines in the known_hosts contents are provided in a valid sshd(8) format') from ex
 
 
-def __get_data_from_key_or_file(key, filepath):
+def get_data_from_key_or_file(key, filepath):
     if key != '' and filepath != '':
         raise MutuallyExclusiveArgumentError(
             'Error! Both textual key and key filepath cannot be provided',
             'Try providing the file parameter without providing the plaintext parameter')
     data = ''
     if filepath != '':
-        data = __read_key_file(filepath)
+        data = read_key_file(filepath)
     elif key != '':
         data = key
     return data
 
 
-def __read_key_file(path):
+def read_key_file(path):
     try:
         with open(path, "r") as myfile:  # user passed in filename
             data_list = myfile.readlines()  # keeps newline characters intact
             data_list_len = len(data_list)
             if (data_list_len) <= 0:
                 raise Exception("File provided does not contain any data")
             raw_data = ''.join(data_list)
-        return __to_base64(raw_data)
+        return to_base64(raw_data)
     except Exception as ex:
         raise InvalidArgumentValueError(
             'Error! Unable to read key file specified with: {0}'.format(ex),
             'Verify that the filepath specified exists and contains valid utf-8 data') from ex
 
 
-def __from_base64(base64_str):
+def from_base64(base64_str):
     return base64.b64decode(base64_str)
 
 
-def __to_base64(raw_data):
+def to_base64(raw_data):
     bytes_data = raw_data.encode('utf-8')
     return base64.b64encode(bytes_data).decode('utf-8')