[k8sconfiguration] Parameter Validation and Table Formatting

src/k8sconfiguration/HISTORY.rst

@@ -3,6 +3,10 @@
 Release History
 ===============
 
+0.2.3
+++++++++++++++++++
+* Add parameter regex validation, improve table formatting
+
 0.2.2
 ++++++++++++++++++
 * Update min az CLI version

src/k8sconfiguration/azext_k8sconfiguration/_format.py

@@ -0,0 +1,33 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from collections import OrderedDict
+
+
+def k8sconfiguration_list_table_format(results):
+    formatted_result = []
+    for result in results:
+        formatted_result.append(
+            OrderedDict([
+                ('name', result['name']),
+                ('repositoryUrl', result['repositoryUrl']),
+                ('operatorName', result['operatorInstanceName']),
+                ('operatorNamespace', result['operatorNamespace']),
+                ('scope', result['operatorScope']),
+                ('provisioningState', result['provisioningState'])
+            ])
+        )
+    return formatted_result
+
+
+def k8sconfiguration_show_table_format(result):
+    return OrderedDict([
+        ('name', result['name']),
+        ('repositoryUrl', result['repositoryUrl']),
+        ('operatorName', result['operatorInstanceName']),
+        ('operatorNamespace', result['operatorNamespace']),
+        ('scope', result['operatorScope']),
+        ('provisioningState', result['provisioningState'])
+    ])

src/k8sconfiguration/azext_k8sconfiguration/_params.py

@@ -13,26 +13,35 @@
 )
 
 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_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'])
+        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',
@@ -42,7 +51,7 @@ def load_arguments(self, _):
         c.argument('operator_params',
                    help='Parameters for the Operator')
         c.argument('ssh_private_key',
-                   help='Specify private ssh key for private repository sync (either base64 encoded or raw)')
+                   help='Specify base64-encoded private ssh key for private repository sync')
         c.argument('ssh_private_key_file',
                    help='Specify filepath to private ssh key for private repository sync')
         c.argument('https_user',
@@ -54,9 +63,11 @@ 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' ''')
 

src/k8sconfiguration/azext_k8sconfiguration/_validators.py

@@ -3,11 +3,44 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
+import re
 from azure.cli.core.azclierror import InvalidArgumentValueError
 
 
+# Parameter-Level Validation
 def validate_configuration_type(configuration_type):
     if configuration_type.lower() != 'sourcecontrolconfiguration':
         raise InvalidArgumentValueError(
             'Invalid configuration-type',
             'Try specifying the valid value "sourceControlConfiguration"')
+
+
+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)
+
+
+# Create Parameter Validation
+def validate_configuration_name(configuration_name):
+    __validate_k8s_name(configuration_name, "--name", 63)
+
+
+# Helper
+def __validate_k8s_name(param_value, param_name, max_len):
+    if len(param_value) > max_len:
+        raise InvalidArgumentValueError(
+            'Error! Invalid {0}'.format(param_name),
+            'Parameter {0} can be a maximum of {1} characters'.format(param_name, max_len))
+    if not re.match(r'^[a-z0-9]([-a-z0-9]*[a-z0-9])?$', param_value):
+        if param_value[0] == "-" or param_value[-1] == "-":
+            raise InvalidArgumentValueError(
+                'Error! Invalid {0}'.format(param_name),
+                'Parameter {0} cannot begin or end with a hyphen'.format(param_name))
+        raise InvalidArgumentValueError(
+            'Error! Invalid {0}'.format(param_name),
+            'Parameter {0} can only contain lowercase alphanumeric characters and hyphens'.format(param_name))

src/k8sconfiguration/azext_k8sconfiguration/commands.py

@@ -6,6 +6,7 @@
 # pylint: disable=line-too-long
 from azure.cli.core.commands import CliCommandType
 from azext_k8sconfiguration._client_factory import (cf_k8sconfiguration, cf_k8sconfiguration_operation)
+from ._format import k8sconfiguration_show_table_format, k8sconfiguration_list_table_format
 
 
 def load_command_table(self, _):
@@ -20,5 +21,5 @@ def load_command_table(self, _):
         g.custom_command('create', 'create_k8sconfiguration')
         g.custom_command('update', 'update_k8sconfiguration')
         g.custom_command('delete', 'delete_k8sconfiguration', confirmation=True)
-        g.custom_command('list', 'list_k8sconfiguration')
-        g.custom_show_command('show', 'show_k8sconfiguration')
+        g.custom_command('list', 'list_k8sconfiguration', table_transformer=k8sconfiguration_list_table_format)
+        g.custom_show_command('show', 'show_k8sconfiguration', table_transformer=k8sconfiguration_show_table_format)

src/k8sconfiguration/azext_k8sconfiguration/custom.py

@@ -17,6 +17,7 @@
 from azext_k8sconfiguration.vendored_sdks.models import SourceControlConfiguration
 from azext_k8sconfiguration.vendored_sdks.models import HelmOperatorProperties
 from azext_k8sconfiguration.vendored_sdks.models import ErrorResponseException
+from ._validators import validate_configuration_name
 
 logger = get_logger(__name__)
 
@@ -62,6 +63,9 @@ def create_k8sconfiguration(client, resource_group_name, cluster_name, name, rep
     """Create a new Kubernetes Source Control Configuration.
 
     """
+    # Validate configuration name
+    validate_configuration_name(name)
+
     # Determine ClusterRP
     cluster_rp = __get_cluster_type(cluster_type)
 
@@ -77,16 +81,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 +137,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 +162,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 +187,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 +221,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 +252,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 +274,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 +297,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')