Validate that zowe.config.json file matches schema

CHANGELOG.md

@@ -4,8 +4,8 @@ All notable changes to the Zowe Client Python SDK will be documented in this fil
 
 ## Recent Changes
 
+- Feature: Added method to load profile properties from environment variables
+- Feature: Added a CredentialManager class to securely retrieve values from credentials and manage multiple credential entries on Windows [#134](https://github.com/zowe/zowe-client-python-sdk/issues/134)
 - Bugfix: Fixed issue for datasets and jobs with special characters in URL [#211] (https://github.com/zowe/zowe-client-python-sdk/issues/211)
 - Bugfix: Fixed exception handling in session.py [#213] (https://github.com/zowe/zowe-client-python-sdk/issues/213)
-
-- Feature: Added a CredentialManager class to securely retrieve values from credentials and manage multiple credential entries on Windows [#134](https://github.com/zowe/zowe-client-python-sdk/issues/134)
-- Feature: Added method to load profile properties from environment variables
+- BugFix: Validation of zowe.config.json file matching the schema [#192](https://github.com/zowe/zowe-client-python-sdk/issues/192)

requirements.txt

@@ -6,7 +6,7 @@ coverage==5.4
 flake8==3.8.4
 idna==2.10
 importlib-metadata==3.6.0
-jsonschema==4.14.0
+jsonschema==4.17.3
 keyring
 lxml==4.9.3
 mccabe==0.6.1

src/core/zowe/core_for_zowe_sdk/config_file.py

@@ -20,6 +20,8 @@
 
 import commentjson
 
+from .constants import constants
+from .validators import validate_config_json
 from .credential_manager import CredentialManager
 from .custom_warnings import (
     ProfileNotFoundWarning,
@@ -72,8 +74,9 @@
     _location: Optional[str] = None
     profiles: Optional[dict] = None
     defaults: Optional[dict] = None
-    secure_props: Optional[dict] = None
     schema_property: Optional[dict] = None
+    secure_props: Optional[dict] = None
+    jsonc: Optional[dict] = None
     _missing_secure_props: list = field(default_factory=list)
 
     @property
@@ -99,7 +102,7 @@
 
     @property
     def schema_path(self) -> Optional[str]:
-        self.schema_property
+        return self.schema_property
 
     @location.setter
     def location(self, dirname: str) -> None:
@@ -108,7 +111,10 @@
         else:
             raise FileNotFoundError(f"given path {dirname} is not valid")
 
-    def init_from_file(self) -> None:
+    def init_from_file(
+        self,
+        validate_schema: Optional[bool] = True,
+    ) -> None:
         """
         Initializes the class variable after
         setting filepath (or if not set, autodiscover the file)
@@ -120,25 +126,51 @@
             profile_jsonc = commentjson.load(fileobj)
 
         self.profiles = profile_jsonc.get("profiles", {})
-        self.defaults = profile_jsonc.get("defaults", {})
         self.schema_property = profile_jsonc.get("$schema", None)
+        self.defaults = profile_jsonc.get("defaults", {})
+        self.jsonc = profile_jsonc
 
+        if self.schema_property and validate_schema:
+            self.validate_schema()
         # loading secure props is done in load_profile_properties
         # since we want to try loading secure properties only when
         # we know that the profile has saved properties
         # self.load_secure_props()
 
+    def validate_schema(
+        self
+    ) -> None:
+        """
+        Get the $schema_property from the config and load the schema
+
+        Returns
+        -------
+        file_path to the $schema property
+        """
+
+        path_schema_json = None
+
+        path_schema_json = self.schema_path
+        if path_schema_json is None:    # check if the $schema property is not defined
+            warnings.warn(
+                f"$schema property could not found"
+            )
+
+        # validate the $schema property
+        if path_schema_json:
+            validate_config_json(self.jsonc, path_schema_json, cwd = self.location)
+
     def schema_list(
         self,
     ) -> list:
         """
         Loads the schema properties
         in a sorted order according to the priority
-        
+
         Returns
         -------
         Dictionary
-        
+
             Returns the profile properties from schema (prop: value)
         """
 
@@ -149,23 +181,23 @@
         if schema.startswith("https://") or schema.startswith("http://"):
             schema_json = requests.get(schema).json()
 
+        elif os.path.isfile(schema) or schema.startswith("file://"):
+            with open(schema.replace("file://", "")) as f:
+                schema_json = json.load(f)
+
         elif not os.path.isabs(schema):
             schema = os.path.join(self.location, schema)
             with open(schema) as f:
                 schema_json = json.load(f)
-
-        elif os.path.isfile(schema):
-            with open(schema) as f:
-                schema_json = json.load(f)
         else:
             return []
 
         profile_props:dict = {}
         schema_json = dict(schema_json)
-        
+
         for props in schema_json['properties']['profiles']['patternProperties']["^\\S*$"]["allOf"]:
             props = props["then"]
-            
+
             while "properties" in props:
                 props = props.pop("properties")
                 profile_props = props
@@ -176,6 +208,7 @@
         self,
         profile_name: Optional[str] = None,
         profile_type: Optional[str] = None,
+        validate_schema: Optional[bool] = True,
     ) -> Profile:
         """
         Load given profile including secure properties and excluding values from base profile
@@ -185,7 +218,7 @@
             Returns a namedtuple called Profile
         """
         if self.profiles is None:
-            self.init_from_file()
+            self.init_from_file(validate_schema)
 
         if profile_name is None and profile_type is None:
             raise ProfileNotFound(
@@ -270,7 +303,7 @@
             profile_name=profile_type,
             error_msg=f"No profile with matching profile_type '{profile_type}' found",
         )
-        
+
     def find_profile(self, path: str, profiles: dict):
         """
         Find a profile at a specified location from within a set of nested profiles

src/core/zowe/core_for_zowe_sdk/credential_manager.py

@@ -10,8 +10,8 @@
 Copyright Contributors to the Zowe Project.
 """
 import sys
-import warnings 
-import base64 
+import warnings
+import base64
 import logging
 from typing import Optional
 import commentjson
@@ -30,7 +30,7 @@
 class CredentialManager:
     secure_props = {}
 
-    
+
 
     @staticmethod
     def load_secure_props() -> None:
@@ -51,7 +51,7 @@
             secret_value = CredentialManager._retrieve_credential(service_name)
             # Handle the case when secret_value is None
             if secret_value is None:
-                return 
+                return
 
         except Exception as exc:
             raise SecureProfileLoadFailed(
@@ -63,9 +63,9 @@
         secure_config_json = commentjson.loads(base64.b64decode(secure_config).decode())
         # update the secure props
         CredentialManager.secure_props = secure_config_json
-        
-    
-    @staticmethod    
+
+
+    @staticmethod
     def _retrieve_credential(service_name: str) -> Optional[str]:
         """
         Retrieve the credential from the keyring or storage.
@@ -96,7 +96,7 @@
                     encoded_credential += temp_value
                 index += 1
                 temp_value = keyring.get_password(f"{service_name}-{index}", f"{constants['ZoweAccountName']}-{index}")
-                
+
         if is_win32:
             try:
                 encoded_credential = encoded_credential.encode('utf-16le').decode()
@@ -106,10 +106,10 @@
 
             if encoded_credential is not None and encoded_credential.endswith("\0"):
                 encoded_credential = encoded_credential[:-1]
-        
-        return encoded_credential        
-        
-    
+
+        return encoded_credential
+
+
     @staticmethod
     def delete_credential(service_name: str, account_name: str) -> None:
         """
@@ -125,7 +125,7 @@
         -------
         None
         """
-        
+
         try:
             keyring.delete_password(service_name, account_name)
         except keyring.errors.PasswordDeleteError:
@@ -143,7 +143,7 @@
                     break
                 index += 1
 
-    
+
     @staticmethod
     def save_secure_props()-> None:
         """
@@ -154,16 +154,16 @@
         """
         if not HAS_KEYRING:
             return
-            
+
         service_name = constants["ZoweServiceName"]
         credential =  CredentialManager.secure_props
         # Check if credential is a non-empty string
         if credential:
             is_win32 = sys.platform == "win32"
-            
-            encoded_credential = base64.b64encode(commentjson.dumps(credential).encode()).decode() 
+
+            encoded_credential = base64.b64encode(commentjson.dumps(credential).encode()).decode()
             if is_win32:
-                service_name += "/" + constants["ZoweAccountName"] 
+                service_name += "/" + constants["ZoweAccountName"]
                 # Delete the existing credential
                 CredentialManager.delete_credential(service_name , constants["ZoweAccountName"])
             # Check if the encoded credential exceeds the maximum length for win32
@@ -177,9 +177,9 @@
                     password=(chunk + '\0' *(len(chunk)%2)).encode().decode('utf-16le')
                     field_name = f"{constants['ZoweAccountName']}-{index}"
                     keyring.set_password(f"{service_name}-{index}", field_name, password)
-                    
+
             else:
                 # Credential length is within the maximum limit or not on win32, set it as a single keyring entry
                 keyring.set_password(
-                    service_name, constants["ZoweAccountName"], 
+                    service_name, constants["ZoweAccountName"],
                     encoded_credential)